// Package zset contains the long-running Redis ZSet task consumer. It is an
// independent runtime role; wiring it into the process is intentionally left
// to the application bootstrap.
package zset

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"strings"
	"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/fullsync/supply"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
	queue "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
)

var (
	ErrAlreadyRunning = errors.New("zset consumer is already running")
	ErrLockLost       = errors.New("zset consumer lock was lost")
	ErrLeaseLost      = errors.New("zset task lease was lost")
)

// Handler is the boundary to the independent Chan gateway and result store.
type Handler interface {
	Query(context.Context, queue.Task) Result
	Apply(context.Context, []Result) error
}

type Result struct {
	Task    queue.Task
	UsageMB float64
	// CardStatus carries status-sync results without overloading the traffic
	// usage field. It is additive so the existing monthly usage handler keeps
	// its original contract.
	CardStatus *int
	// Imei carries the IMEI-sync value. It is separate from CardStatus because
	// an IMEI is a string and because "synced with an empty value" is a
	// meaningful state that Success alone cannot express.
	Imei string
	// ImeiSynced reports that the carrier answered successfully, so the card
	// was covered by this polling pass. It is independent of Success: a
	// successful-but-empty answer sets ImeiSynced without Success, which is how
	// the write path records coverage while leaving the stored value untouched.
	ImeiSynced   bool
	Code         int
	ErrCode      string
	Message      string
	ErrorClass   string
	Success      bool
	Retryable    bool
	BlockChannel bool
	// Covered marks a terminal answer that says something definitive about the
	// card but yields no value, such as "the carrier holds no record of this
	// card" (code 14000) or "the carrier returned an empty IMEI". The run was
	// covered for that card, so it is neither retried nor counted as a failure;
	// Success stays false because nothing may be written to the stored value.
	Covered bool
}

// Config controls the bounded scheduling window. Values are deliberately
// explicit so production limits can be selected after the capacity PoC.
type Config struct {
	SupplyObservation  *supply.Tracker
	LockKey            string
	LockTTL            time.Duration
	PollInterval       time.Duration
	LeaseTTL           time.Duration
	TaskTimeout        time.Duration
	ClaimBatch         int
	MobileConcurrency  int
	TelecomConcurrency int
	GenericConcurrency int
	// GenericQPS is the per-channel pacing rate used for unknown templates.
	GenericQPS float64
	// TemplateQPS overrides the per-channel pacing rate for channel templates.
	// A nil map uses the built-in template defaults.
	TemplateQPS       map[string]float64
	RunQueueEnabled   bool
	RoundRetryEnabled bool
	MaxRounds         int
	FailureRetention  time.Duration
	// FailurePoolAlertThreshold emits a warning when one run's terminal failure
	// pool exceeds this many entries. It is an observability guard, not a
	// limiter: the run is never aborted, because a genuinely broken upstream
	// must still leave a complete, reviewable failure list. Zero disables it.
	FailurePoolAlertThreshold int64
	// ShouldRetryNextRound selects failures that are deferred into the next
	// Run round. A nil predicate preserves the monthly-usage mobile_busy rule.
	ShouldRetryNextRound func(Result) bool
	OnFailure            func(context.Context, Result)
	// BlockRun optionally provides a run-scoped channel blocker. The legacy
	// monthly consumer continues to use Store.BlockRun by default.
	BlockRun func(context.Context, string, string) (int64, error)
	// Status synchronization counts blocked queued tasks as terminal failures.
	BlockedTasksAsFailures bool
	EventPrefix            string
}

const (
	telecomTempKey            = "sim_dx_v1"
	mobileTempKey             = "sim_yd_v5"
	mobileMockKey             = "sim_yd_mock_v5"
	sharedTengyuTempKey       = "sim_onelink_tengyu_v1"
	sharedXuzhouMobileTempKey = "sim_onelink_xuzhou_mobile_v1"
	sharedLelianTempKey       = "sim_onelink_lelian_v1"
	sharedQianxunTempKey      = "sim_onelink_qianxun_v1"
	sharedHenanTempKey        = "sim_onelink_henan_xinjing_v1"
	sharedDaiheTempKey        = "sim_onelink_daihe_v1"
	sharedBeixinTempKey       = "sim_onelink_beixin_v1"
	// lanshang_8000 / lanshang_8003 是蓝尚的 OneLink 共享 Token 模板。它们
	// 虽然名字不带 sim_onelink_ 前缀，但走的是同一套移动 OneLink 协议，
	// 必须归入 carrierMobile：否则会落到 carrierGeneric 的单通道 1.5 QPS，
	// 全量同步一张卡要 0.67 秒，6060 张卡要跑 1 小时以上。
	sharedLanshang8000TempKey = "lanshang_8000"
	sharedLanshang8003TempKey = "lanshang_8003"
	telecomQPS                = 25.0
	mobileQPS                 = 30.0
	defaultGenericQPS         = 1.5
)

type carrier string

const (
	carrierMobile  carrier = "mobile"
	carrierTelecom carrier = "telecom"
	carrierGeneric carrier = "generic"
)

func (c *Config) setDefaults() {
	if c.LockKey == "" {
		c.LockKey = "flow:consumer:active"
	}
	if c.LockTTL <= 0 {
		c.LockTTL = time.Minute
	}
	if c.PollInterval <= 0 {
		c.PollInterval = time.Second
	}
	if c.LeaseTTL <= 0 {
		c.LeaseTTL = time.Minute
	}
	if c.TaskTimeout <= 0 {
		c.TaskTimeout = 30 * time.Second
	}
	if c.ClaimBatch <= 0 {
		c.ClaimBatch = 100
	}
	if c.MobileConcurrency <= 0 {
		c.MobileConcurrency = 30
	}
	if c.TelecomConcurrency <= 0 {
		c.TelecomConcurrency = 10
	}
	if c.GenericConcurrency <= 0 {
		c.GenericConcurrency = 1
	}
	if c.GenericQPS <= 0 {
		c.GenericQPS = defaultGenericQPS
	}
	if len(c.TemplateQPS) > 0 {
		normalized := make(map[string]float64, len(c.TemplateQPS))
		for key, qps := range c.TemplateQPS {
			key = strings.ToLower(strings.TrimSpace(key))
			if key != "" && qps > 0 {
				normalized[key] = qps
			}
		}
		c.TemplateQPS = normalized
	}
	if c.RoundRetryEnabled {
		c.RunQueueEnabled = true
	}
	if c.RunQueueEnabled && !c.RoundRetryEnabled {
		c.MaxRounds = 1
	} else if c.MaxRounds <= 0 {
		c.MaxRounds = 5
	}
	if c.EventPrefix == "" {
		c.EventPrefix = "monthly_usage"
	}
	if c.FailureRetention <= 0 {
		c.FailureRetention = 30 * 24 * time.Hour
	}
}

// Consumer periodically enumerates registered channels and handles a bounded
// claim window for each one. It does not scan the database or implement any
// carrier protocol.
type Consumer struct {
	store           *queue.Store
	handler         Handler
	config          Config
	owner           string
	pools           *channelPools
	pacer           *channelPacer
	runningMu       sync.Mutex
	runningChannels map[string]struct{}
	channelGroup    sync.WaitGroup
	summaryMu       sync.Mutex
	roundSummaries  map[string]*roundSummary
}

type roundSummary struct {
	runID               string
	round               int
	inputCount          int
	successCount        int
	failedCount         int
	deferredCount       int
	terminalFailedCount int
	startedAt           time.Time
	lastBatchAt         time.Time
	batchCount          int
}

func New(store *queue.Store, handler Handler, config Config) (*Consumer, error) {
	if store == nil {
		return nil, fmt.Errorf("zset store is nil")
	}
	if handler == nil {
		return nil, fmt.Errorf("zset handler is nil")
	}
	config.setDefaults()
	if config.BlockRun == nil {
		config.BlockRun = store.BlockRun
	}
	if config.LeaseTTL <= config.TaskTimeout {
		return nil, fmt.Errorf("lease ttl must exceed task timeout")
	}
	return &Consumer{
		store:           store,
		handler:         handler,
		config:          config,
		owner:           newOwner(),
		pools:           newChannelPools(config.MobileConcurrency, config.TelecomConcurrency, config.GenericConcurrency),
		pacer:           newChannelPacer(),
		runningChannels: make(map[string]struct{}),
		roundSummaries:  make(map[string]*roundSummary),
	}, nil
}

// Run obtains a single active-consumer lease and processes until ctx is done.
// A second active instance receives ErrAlreadyRunning and does not claim work.
func (c *Consumer) Run(ctx context.Context) (err error) {
	if ctx == nil {
		ctx = context.Background()
	}
	startedAt := time.Now()
	acquired, err := c.store.AcquireLock(ctx, c.config.LockKey, c.owner, c.config.LockTTL)
	if err != nil {
		g.Log().Errorf(ctx, "event=%s_consumer status=failed stage=lock durationMs=%d error=%q", c.config.EventPrefix, time.Since(startedAt).Milliseconds(), err.Error())
		return err
	}
	if !acquired {
		g.Log().Warningf(ctx, "event=%s_consumer status=skipped reason=already_running durationMs=%d", c.config.EventPrefix, time.Since(startedAt).Milliseconds())
		return ErrAlreadyRunning
	}
	if observer := c.config.SupplyObservation; observer != nil {
		observer.ConsumerActive.Store(true)
		defer observer.ConsumerActive.Store(false)
	}
	g.Log().Infof(ctx, "event=%s_consumer status=started lockKey=%s", c.config.EventPrefix, c.config.LockKey)

	runCtx, cancel := context.WithCancel(ctx)
	defer func() {
		cancel()
		c.channelGroup.Wait()
		releaseCtx, releaseCancel := context.WithTimeout(context.Background(), time.Second)
		defer releaseCancel()
		_, _ = c.store.ReleaseLock(releaseCtx, c.config.LockKey, c.owner)
		status := "stopped"
		if err != nil {
			status = "failed"
		}
		message := fmt.Sprintf("event=%s_consumer status=%s durationMs=%d error=%q", c.config.EventPrefix, status, time.Since(startedAt).Milliseconds(), errorMessage(err))
		if err != nil {
			g.Log().Error(ctx, message)
			return
		}
		g.Log().Info(ctx, message)
	}()

	renewErr := make(chan error, 1)
	go c.renewLock(runCtx, cancel, renewErr)

	for {
		_, err := c.processOnce(runCtx)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			select {
			case lockErr := <-renewErr:
				return lockErr
			default:
			}
			return err
		}
		timer := time.NewTimer(c.config.PollInterval)
		select {
		case <-runCtx.Done():
			if ctx.Err() != nil {
				return nil
			}
			select {
			case err := <-renewErr:
				return err
			default:
				return runCtx.Err()
			}
		case err := <-renewErr:
			return err
		case <-timer.C:
		}
	}
}

func (c *Consumer) renewLock(ctx context.Context, cancel context.CancelFunc, errors chan<- error) {
	interval := c.config.LockTTL / 3
	if interval <= 0 {
		interval = time.Millisecond
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			ok, err := c.store.RenewLock(ctx, c.config.LockKey, c.owner, c.config.LockTTL)
			if err != nil {
				select {
				case errors <- err:
				default:
				}
				cancel()
				return
			}
			if !ok {
				select {
				case errors <- ErrLockLost:
				default:
				}
				cancel()
				return
			}
		}
	}
}

func (c *Consumer) processOnce(ctx context.Context) (bool, error) {
	started := false
	if c.config.RunQueueEnabled {
		runs, err := c.store.ActiveRuns(ctx)
		if err != nil {
			return false, err
		}
		for _, runID := range runs {
			// The transition is guarded by the Run metadata counters, so it is
			// safe for every poller to attempt it.
			meta, err := c.store.GetRunMeta(ctx, runID)
			if err != nil {
				return false, err
			}
			if c.config.RoundRetryEnabled {
				maxRounds := c.config.MaxRounds
				if meta.MaxRound > 0 {
					maxRounds = meta.MaxRound
				}
				advanced, err := c.store.AdvanceRunRound(ctx, runID, maxRounds)
				if err != nil {
					return false, err
				}
				if advanced {
					c.flushRoundSummary(ctx, runID, meta.CurrentRound)
				}
			}
			channels, err := c.store.RunChannels(ctx, runID)
			if err != nil {
				return false, err
			}
			for _, chanKey := range channels {
				started = c.startRunChannel(ctx, runID, chanKey) || started
			}
			if err := c.completeRun(ctx, runID); err != nil {
				return false, err
			}
		}
	}
	channels, err := c.store.ActiveChannels(ctx)
	if err != nil {
		return false, err
	}
	if len(channels) == 0 {
		return started, nil
	}

	for _, chanKey := range channels {
		started = c.startChannel(ctx, chanKey) || started
	}
	return started, nil
}

func (c *Consumer) startRunChannel(ctx context.Context, runID, chanKey string) bool {
	key := runID + "\x00" + chanKey
	c.runningMu.Lock()
	if _, exists := c.runningChannels[key]; exists {
		c.runningMu.Unlock()
		return false
	}
	c.runningChannels[key] = struct{}{}
	c.channelGroup.Add(1)
	c.runningMu.Unlock()

	go func() {
		defer c.channelGroup.Done()
		defer func() {
			c.runningMu.Lock()
			delete(c.runningChannels, key)
			c.runningMu.Unlock()
		}()
		startedAt := time.Now()
		for ctx.Err() == nil {
			processed, err := c.processRunChannel(ctx, runID, chanKey)
			if err != nil {
				g.Log().Errorf(ctx, "event=%s_batch status=failed stage=claim_or_process runId=%s chanKey=%s durationMs=%d error=%q", c.config.EventPrefix, runID, chanKey, time.Since(startedAt).Milliseconds(), err.Error())
				return
			}
			if !processed {
				return
			}
		}
	}()
	return true
}

func (c *Consumer) processRunChannel(ctx context.Context, runID, chanKey string) (bool, error) {
	now := time.Now()
	claimed, err := c.store.ClaimRun(ctx, runID, chanKey, now, now.Add(c.config.LeaseTTL), c.config.ClaimBatch)
	if err != nil {
		return false, err
	}
	if len(claimed) == 0 {
		if observer := c.config.SupplyObservation; observer != nil {
			observer.Empty(runID, chanKey)
		}
		return false, nil
	}
	if observer := c.config.SupplyObservation; observer != nil {
		observer.Processing(runID, chanKey, 1)
		defer observer.Processing(runID, chanKey, -1)
	}
	return true, c.handleRunClaims(ctx, claimed)
}

func (c *Consumer) handleRunClaims(ctx context.Context, claimed []queue.ClaimedTask) error {
	if len(claimed) == 0 {
		return nil
	}
	startedAt := time.Now()
	channelCarrier, qps := policyForTemplate(claimed[0].Task.TempKey, c.config)
	batchCtx, cancelBatch := context.WithCancel(ctx)
	queryCtx, cancelQueries := context.WithCancel(batchCtx)
	defer cancelQueries()
	renewErr := make(chan error, 1)
	renewDone := make(chan struct{})
	go c.renewBatchLeases(batchCtx, cancelBatch, claimed, renewErr, renewDone)
	defer func() { cancelBatch(); <-renewDone }()

	var group sync.WaitGroup
	resultsCh := make(chan Result, len(claimed))
	for _, task := range claimed {
		task := task
		group.Add(1)
		go func() {
			defer group.Done()
			result := c.queryClaim(queryCtx, task, channelCarrier, qps)
			if result.BlockChannel {
				cancelQueries()
			}
			resultsCh <- result
		}()
	}
	group.Wait()
	close(resultsCh)
	results := make([]Result, 0, len(claimed))
	blockChannel := false
	for result := range resultsCh {
		results = append(results, result)
		blockChannel = blockChannel || result.BlockChannel
	}
	if err := currentBatchError(batchCtx, renewErr); err != nil {
		return err
	}
	for _, result := range results {
		if !result.Success && !result.Covered && c.config.OnFailure != nil {
			c.config.OnFailure(ctx, result)
		}
	}
	if err := c.handler.Apply(batchCtx, results); err != nil {
		return gerror.Wrap(err, "apply run result batch")
	}
	if err := currentBatchError(batchCtx, renewErr); err != nil {
		return err
	}

	meta, err := c.store.GetRunMeta(batchCtx, claimed[0].Task.RunID)
	if err != nil {
		return err
	}
	maxRounds := c.config.MaxRounds
	if meta.MaxRound > 0 {
		maxRounds = meta.MaxRound
	}
	finalizedClaims := make([]queue.ClaimedTask, 0, len(results))
	outcomes := make([]queue.TaskOutcome, 0, len(results))
	deferred := 0
	for _, result := range results {
		// A telecom channel-limit response is handled by BlockRunScoped below.
		// Leave the triggering lease in place so the same atomic block operation
		// marks it (and all other queued/leased channel tasks) channel_blocked;
		// it must not become a mobile round retry or a card-level final failure.
		if result.BlockChannel {
			continue
		}
		roundRetry := result.Retryable && c.shouldRetryNextRound(result)
		if result.Retryable && !roundRetry && !result.Success && !result.BlockChannel {
			// Infrastructure and lease-recoverable failures intentionally stay
			// leased; the existing lease expiry path will recover them.
			continue
		}
		claimedTask := claimedTaskForResult(claimed, result)
		if claimedTask.Task.TaskID == "" {
			continue
		}
		outcome := queue.TaskOutcome{
			Success:      result.Success,
			Retryable:    false,
			ErrorClass:   result.ErrorClass,
			FailureType:  result.ErrorClass,
			Code:         result.Code,
			ErrCode:      result.ErrCode,
			Message:      result.Message,
			BlockChannel: result.BlockChannel,
			Covered:      result.Covered,
		}
		if c.config.RoundRetryEnabled && roundRetry && meta.CurrentRound < maxRounds {
			outcome.Retry = true
			deferred++
		}
		finalizedClaims = append(finalizedClaims, claimedTask)
		outcomes = append(outcomes, outcome)
	}
	finalized := 0
	if len(finalizedClaims) > 0 {
		count, finalizeErr := c.store.FinalizeRunTasks(batchCtx, finalizedClaims, outcomes, maxRounds)
		if finalizeErr != nil {
			return finalizeErr
		}
		if count != len(finalizedClaims) {
			return ErrLeaseLost
		}
		finalized = count
	}
	if blockChannel {
		if c.config.BlockedTasksAsFailures {
			for _, result := range results {
				if !result.BlockChannel {
					continue
				}
				removed, err := c.store.BlockRunScopedAsFailed(batchCtx, claimedTaskForResult(claimed, result), queue.TaskOutcome{Code: result.Code, ErrCode: result.ErrCode, Message: result.Message})
				if err != nil {
					return err
				}
				if removed < 0 {
					return ErrLeaseLost
				}
				c.recordRoundCounts(claimed[0].Task.RunID, meta.CurrentRound, int(removed), 0, 0, int(removed), startedAt)
				break
			}
		} else if _, err := c.store.BlockRunScoped(batchCtx, claimed[0].Task.RunID, claimed[0].Task.ChanKey); err != nil {
			return err
		}
	}
	c.recordRoundBatch(claimed[0].Task.RunID, meta.CurrentRound, finalizedClaims, outcomes, startedAt)
	if err := c.completeRun(batchCtx, claimed[0].Task.RunID); err != nil {
		return err
	}
	g.Log().Infof(ctx, "event=%s_round_batch status=completed runId=%s queryMonth=%s chanKey=%s tempKey=%s round=%d claimed=%d success=%d covered=%d failed=%d deferred=%d finalized=%d durationMs=%d", c.config.EventPrefix, claimed[0].Task.RunID, claimed[0].Task.QueryMonth, claimed[0].Task.ChanKey, claimed[0].Task.TempKey, meta.CurrentRound, len(claimed), countSuccessful(results), countCovered(results), len(results)-countSuccessful(results)-countCovered(results), deferred, finalized, time.Since(startedAt).Milliseconds())
	return nil
}

func (c *Consumer) recordRoundBatch(runID string, round int, claimed []queue.ClaimedTask, outcomes []queue.TaskOutcome, startedAt time.Time) {
	if len(claimed) == 0 || len(claimed) != len(outcomes) {
		return
	}
	success, failed, deferred := 0, 0, 0
	for _, outcome := range outcomes {
		if outcome.Success || outcome.Covered {
			// A covered task answered definitively and produced no failure, so
			// it closes the round like a success. Keeping it out of both buckets
			// would break the enqueued = success + nextRound + failed identity
			// that the run archive validates.
			success++
			continue
		}
		failed++
		if outcome.Retry {
			deferred++
		}
	}
	c.recordRoundCounts(runID, round, len(claimed), success, deferred, failed-deferred, startedAt)
}

func (c *Consumer) recordRoundCounts(runID string, round, input, success, deferred, terminalFailed int, startedAt time.Time) {
	if input <= 0 {
		return
	}
	key := runID + "\x00" + fmt.Sprint(round)
	now := time.Now()
	c.summaryMu.Lock()
	summary := c.roundSummaries[key]
	if summary == nil {
		summary = &roundSummary{runID: runID, round: round, startedAt: startedAt}
		c.roundSummaries[key] = summary
	}
	if startedAt.Before(summary.startedAt) {
		summary.startedAt = startedAt
	}
	summary.inputCount += input
	summary.successCount += success
	summary.failedCount += deferred + terminalFailed
	summary.deferredCount += deferred
	summary.terminalFailedCount += terminalFailed
	summary.batchCount++
	summary.lastBatchAt = now
	c.summaryMu.Unlock()
}

func (c *Consumer) flushRoundSummary(ctx context.Context, runID string, round int) {
	key := runID + "\x00" + fmt.Sprint(round)
	c.summaryMu.Lock()
	summary := c.roundSummaries[key]
	c.summaryMu.Unlock()
	if summary == nil {
		return
	}
	if summary.lastBatchAt.IsZero() {
		summary.lastBatchAt = time.Now()
	}
	completedAt := summary.lastBatchAt
	stats, err := c.store.GetRoundStats(ctx, runID)
	if err != nil {
		g.Log().Warningf(ctx, "event=%s_round_summary status=failed runId=%s round=%d stage=read_existing error=%q", c.config.EventPrefix, runID, round, err.Error())
		return
	}
	startedAt := summary.startedAt
	entry := queue.RoundStats{
		RunID:          runID,
		Round:          round,
		EnqueuedCount:  int64(summary.inputCount),
		NextRoundCount: int64(summary.deferredCount),
		SuccessCount:   int64(summary.successCount),
		FailedCount:    int64(summary.terminalFailedCount),
		BatchCount:     int64(summary.batchCount),
		StartedAt:      &startedAt,
		CompletedAt:    &completedAt,
	}
	found := false
	for i := range stats {
		if stats[i].Round == round {
			stats[i] = entry
			found = true
			break
		}
	}
	if !found {
		stats = append(stats, entry)
	}
	if err := c.store.SaveRoundStats(ctx, runID, stats); err != nil {
		g.Log().Warningf(ctx, "event=%s_round_summary status=failed runId=%s round=%d stage=persist error=%q", c.config.EventPrefix, runID, round, err.Error())
		return
	}
	c.summaryMu.Lock()
	delete(c.roundSummaries, key)
	c.summaryMu.Unlock()
	g.Log().Infof(ctx, "event=%s_round_summary status=completed runId=%s round=%d inputCount=%d successCount=%d failedCount=%d deferredCount=%d terminalFailedCount=%d startedAt=%s completedAt=%s durationMs=%d", c.config.EventPrefix, runID, round, summary.inputCount, summary.successCount, summary.failedCount, summary.deferredCount, summary.terminalFailedCount, summary.startedAt.Format(time.RFC3339), summary.lastBatchAt.Format(time.RFC3339), summary.lastBatchAt.Sub(summary.startedAt).Milliseconds())
}

func (c *Consumer) completeRun(ctx context.Context, runID string) error {
	completed, err := c.store.CompleteRunIfIdle(ctx, runID, c.config.FailureRetention)
	if err != nil || !completed {
		return err
	}
	meta, err := c.store.GetRunMeta(ctx, runID)
	if err != nil {
		return gerror.Wrap(err, "read completed run metadata")
	}
	c.flushRoundSummary(ctx, runID, meta.CurrentRound)
	g.Log().Infof(ctx, "event=%s_run status=%s runId=%s successCount=%d failedCount=%d completedAt=%s", c.config.EventPrefix, meta.Status, runID, meta.SuccessCount, meta.FailedCount, meta.CompletedAt.Format(time.RFC3339))
	c.reportFailurePoolSize(ctx, runID, meta)
	return nil
}

// reportFailurePoolSize warns when one run left behind an unusually large
// terminal failure pool.
//
// The pool is the only per-card structure that grows with the card count
// (each FailureRecord is roughly 1KB of JSON), and it is retained for
// FailureRetention — so an upstream-wide outage on a large run is the one
// scenario that can push Redis usage up by hundreds of megabytes. The alert
// exists so that condition is visible in logs instead of being discovered as a
// memory graph later; the run itself is deliberately never truncated.
func (c *Consumer) reportFailurePoolSize(ctx context.Context, runID string, meta queue.RunMeta) {
	threshold := c.config.FailurePoolAlertThreshold
	// Strictly greater than: the documented contract is "exceeds this many
	// entries", so a run sitting exactly on the threshold is not yet an
	// anomaly.
	if threshold <= 0 || meta.FailedCount <= threshold {
		return
	}
	failureKey, err := c.store.FailedKey(runID)
	if err != nil {
		return
	}
	size, err := c.store.RedisClient().Do(ctx, "HLEN", failureKey)
	if err != nil {
		return
	}
	g.Log().Warningf(ctx, "event=%s_failure_pool status=large runId=%s failedCount=%d poolEntries=%d threshold=%d retention=%s action=none_observability_only",
		c.config.EventPrefix, runID, meta.FailedCount, size.Int(), threshold, c.config.FailureRetention)
}

func (c *Consumer) shouldRetryNextRound(result Result) bool {
	if c.config.ShouldRetryNextRound != nil {
		return c.config.ShouldRetryNextRound(result)
	}
	return result.ErrorClass == "mobile_busy"
}

func (c *Consumer) startChannel(ctx context.Context, chanKey string) bool {
	c.runningMu.Lock()
	if _, exists := c.runningChannels[chanKey]; exists {
		c.runningMu.Unlock()
		return false
	}
	c.runningChannels[chanKey] = struct{}{}
	c.channelGroup.Add(1)
	c.runningMu.Unlock()

	go func() {
		defer c.channelGroup.Done()
		defer func() {
			c.runningMu.Lock()
			delete(c.runningChannels, chanKey)
			c.runningMu.Unlock()
		}()
		startedAt := time.Now()
		for ctx.Err() == nil {
			processed, err := c.processChannel(ctx, chanKey)
			if err != nil {
				g.Log().Errorf(ctx, "event=monthly_usage_batch status=failed stage=claim_or_process chanKey=%s durationMs=%d error=%q", chanKey, time.Since(startedAt).Milliseconds(), err.Error())
				return
			}
			if !processed {
				return
			}
		}
	}()
	return true
}

func (c *Consumer) processChannel(ctx context.Context, chanKey string) (bool, error) {
	now := time.Now()
	claimed, err := c.store.Claim(ctx, chanKey, now, now.Add(c.config.LeaseTTL), c.config.ClaimBatch)
	if err != nil {
		return false, err
	}
	if len(claimed) == 0 {
		return false, nil
	}
	return true, c.handleClaims(ctx, claimed)
}

func (c *Consumer) handleClaims(ctx context.Context, claimed []queue.ClaimedTask) error {
	if len(claimed) == 0 {
		return nil
	}
	startedAt := time.Now()
	logBatchFailure := func(stage string, err error) {
		g.Log().Errorf(ctx, "event=monthly_usage_batch status=failed stage=%s runId=%s queryMonth=%s chanKey=%s tempKey=%s claimed=%d durationMs=%d error=%q", stage, claimed[0].Task.RunID, claimed[0].Task.QueryMonth, claimed[0].Task.ChanKey, claimed[0].Task.TempKey, len(claimed), time.Since(startedAt).Milliseconds(), errorMessage(err))
	}
	channelCarrier, qps := policyForTemplate(claimed[0].Task.TempKey, c.config)
	batchCtx, cancelBatch := context.WithCancel(ctx)
	queryCtx, cancelQueries := context.WithCancel(batchCtx)
	defer cancelQueries()
	renewErr := make(chan error, 1)
	renewDone := make(chan struct{})
	go c.renewBatchLeases(batchCtx, cancelBatch, claimed, renewErr, renewDone)
	defer func() {
		cancelBatch()
		<-renewDone
	}()

	var group sync.WaitGroup
	resultsCh := make(chan Result, len(claimed))
	for _, task := range claimed {
		task := task
		group.Add(1)
		go func() {
			defer group.Done()
			result := c.queryClaim(queryCtx, task, channelCarrier, qps)
			if result.BlockChannel {
				cancelQueries()
			}
			resultsCh <- result
		}()
	}
	group.Wait()
	close(resultsCh)
	results := make([]Result, 0, len(claimed))
	blockChannel := false
	for result := range resultsCh {
		results = append(results, result)
		blockChannel = blockChannel || result.BlockChannel
	}
	if err := currentBatchError(batchCtx, renewErr); err != nil {
		logBatchFailure("query_or_lease", err)
		return err
	}
	for _, result := range results {
		if !result.Success && c.config.OnFailure != nil {
			c.config.OnFailure(ctx, result)
		}
	}

	if applyErr := c.handler.Apply(batchCtx, results); applyErr != nil {
		logBatchFailure("apply", applyErr)
		return fmt.Errorf("apply monthly usage result batch: %w", applyErr)
	}
	if err := currentBatchError(batchCtx, renewErr); err != nil {
		logBatchFailure("lease", err)
		return err
	}
	if blockChannel {
		removed, err := c.config.BlockRun(batchCtx, claimed[0].Task.ChanKey, claimed[0].Task.RunID)
		if err != nil {
			logBatchFailure("block", err)
			return err
		}
		g.Log().Errorf(ctx, "event=monthly_usage_channel status=blocked runId=%s queryMonth=%s chanKey=%s tempKey=%s errCode=S00004 claimed=%d removed=%d durationMs=%d", claimed[0].Task.RunID, claimed[0].Task.QueryMonth, claimed[0].Task.ChanKey, claimed[0].Task.TempKey, len(claimed), removed, time.Since(startedAt).Milliseconds())
		return nil
	}
	acknowledged := make([]queue.ClaimedTask, 0, len(claimed))
	for _, result := range results {
		// The legacy channel queue has no round state.  A mobile busy response
		// therefore remains a terminal result there; only Run-scoped consumers
		// can defer it into the next round.  Infrastructure failures retain the
		// historical lease-recovery behaviour.
		if result.Success || !result.Retryable || (result.ErrorClass == "mobile_busy" && !c.config.RoundRetryEnabled) {
			acknowledged = append(acknowledged, claimedTaskForResult(claimed, result))
		}
	}
	if err := c.ackBatch(batchCtx, acknowledged); err != nil {
		logBatchFailure("ack", err)
		return err
	}
	success := countSuccessful(results)
	retryCount := 0
	for _, result := range results {
		if !result.Success && result.Retryable && !result.BlockChannel && !(result.ErrorClass == "mobile_busy" && !c.config.RoundRetryEnabled) {
			retryCount++
		}
	}
	g.Log().Infof(ctx, "event=monthly_usage_batch status=completed runId=%s queryMonth=%s chanKey=%s tempKey=%s claimed=%d success=%d failed=%d retryPending=%d dbUpdated=%d acked=%d durationMs=%d", claimed[0].Task.RunID, claimed[0].Task.QueryMonth, claimed[0].Task.ChanKey, claimed[0].Task.TempKey, len(claimed), success, len(results)-success, retryCount, success, len(acknowledged), time.Since(startedAt).Milliseconds())
	return nil
}

func (c *Consumer) renewBatchLeases(ctx context.Context, cancel context.CancelFunc, claimed []queue.ClaimedTask, errors chan<- error, done chan<- struct{}) {
	defer close(done)
	interval := c.config.LeaseTTL / 3
	if interval <= 0 {
		interval = time.Millisecond
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			var count int
			var err error
			if claimed[0].Round > 0 {
				count, err = c.store.RenewRunBatch(ctx, claimed, time.Now().Add(c.config.LeaseTTL))
			} else {
				count, err = c.store.RenewBatch(ctx, claimed, time.Now().Add(c.config.LeaseTTL))
			}
			if err == nil && count != len(claimed) {
				err = ErrLeaseLost
			}
			if err != nil {
				select {
				case errors <- err:
				default:
				}
				cancel()
				return
			}
		}
	}
}

func currentBatchError(ctx context.Context, renewErr <-chan error) error {
	select {
	case err := <-renewErr:
		return err
	default:
	}
	return ctx.Err()
}

func (c *Consumer) queryClaim(ctx context.Context, claimed queue.ClaimedTask, channelCarrier carrier, qps float64) Result {
	if err := c.pacer.Wait(ctx, claimed.Task.ChanKey, qps); err != nil {
		return Result{Task: claimed.Task, ErrorClass: "canceled", Message: err.Error(), Retryable: true}
	}
	semaphore := c.pools.get(claimed.Task.ChanKey, channelCarrier)
	select {
	case semaphore <- struct{}{}:
	case <-ctx.Done():
		return Result{Task: claimed.Task, ErrorClass: "canceled", Message: ctx.Err().Error(), Retryable: true}
	}
	defer func() { <-semaphore }()
	taskCtx := ctx
	var cancel context.CancelFunc
	if c.config.TaskTimeout > 0 {
		taskCtx, cancel = context.WithTimeout(ctx, c.config.TaskTimeout)
		defer cancel()
	}
	return c.handler.Query(taskCtx, claimed.Task)
}

func claimedTaskForResult(claimed []queue.ClaimedTask, result Result) queue.ClaimedTask {
	for _, item := range claimed {
		if item.Task.TaskID == result.Task.TaskID {
			return item
		}
	}
	return queue.ClaimedTask{}
}

func (c *Consumer) ackBatch(ctx context.Context, claimed []queue.ClaimedTask) error {
	if len(claimed) == 0 {
		return nil
	}
	count, err := c.store.AckBatch(ctx, claimed)
	if err != nil {
		return err
	}
	if count != len(claimed) {
		return ErrLeaseLost
	}
	return nil
}

func countSuccessful(results []Result) int {
	count := 0
	for _, result := range results {
		if result.Success {
			count++
		}
	}
	return count
}

// countCovered counts results that were answered definitively but carry no
// value. They are logged separately from failures because they are neither a
// success (nothing was written) nor a failure (nothing went wrong).
func countCovered(results []Result) int {
	count := 0
	for _, result := range results {
		if !result.Success && result.Covered {
			count++
		}
	}
	return count
}

func errorMessage(err error) string {
	if err == nil {
		return ""
	}
	return err.Error()
}

// channelPools hands each chanKey its own in-flight semaphore. One channel is
// one carrier account, so concurrency must never be shared across channels.
type channelPools struct {
	mu      sync.Mutex
	sems    map[string]chan struct{}
	mobile  int
	telecom int
	generic int
}

func newChannelPools(mobile, telecom, generic int) *channelPools {
	return &channelPools{sems: make(map[string]chan struct{}), mobile: mobile, telecom: telecom, generic: generic}
}

func (p *channelPools) get(chanKey string, channelCarrier carrier) chan struct{} {
	p.mu.Lock()
	defer p.mu.Unlock()
	if sem, ok := p.sems[chanKey]; ok {
		return sem
	}
	size := p.mobile
	if channelCarrier == carrierTelecom {
		size = p.telecom
	} else if channelCarrier == carrierGeneric {
		size = p.generic
	}
	sem := make(chan struct{}, size)
	p.sems[chanKey] = sem
	return sem
}

type channelPacer struct {
	mu   sync.Mutex
	next map[string]time.Time
}

func newChannelPacer() *channelPacer {
	return &channelPacer{next: make(map[string]time.Time)}
}

func (p *channelPacer) Wait(ctx context.Context, chanKey string, qps float64) error {
	delay := p.reserve(chanKey, qps, time.Now())
	if delay <= 0 {
		return nil
	}
	timer := time.NewTimer(delay)
	defer timer.Stop()
	select {
	case <-timer.C:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

func (p *channelPacer) reserve(chanKey string, qps float64, now time.Time) time.Duration {
	if qps <= 0 {
		return 0
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	interval := time.Duration(float64(time.Second) / qps)
	if interval < time.Nanosecond {
		interval = time.Nanosecond
	}
	next := p.next[chanKey]
	if next.Before(now) {
		next = now
	}
	p.next[chanKey] = next.Add(interval)
	return next.Sub(now)
}

func policyForTemplate(tempKey string, config Config) (carrier, float64) {
	tempKey = strings.ToLower(strings.TrimSpace(tempKey))
	qps := func(defaultRate float64) float64 {
		if configured, ok := config.TemplateQPS[tempKey]; ok && configured > 0 {
			return configured
		}
		return defaultRate
	}
	if model.IsTelecomTemplate(tempKey) {
		return carrierTelecom, qps(telecomQPS)
	}
	switch tempKey {
	case mobileTempKey, mobileMockKey, sharedTengyuTempKey, sharedXuzhouMobileTempKey, sharedLelianTempKey, sharedQianxunTempKey, sharedHenanTempKey, sharedDaiheTempKey, sharedBeixinTempKey, sharedLanshang8000TempKey, sharedLanshang8003TempKey:
		return carrierMobile, qps(mobileQPS)
	default:
		genericRate := config.GenericQPS
		if genericRate <= 0 {
			genericRate = defaultGenericQPS
		}
		return carrierGeneric, qps(genericRate)
	}
}

// carrierForOperator is retained as a small compatibility helper for status
// policy tests and callers that only have the channel's operator dictionary.
func carrierForOperator(operator int) (carrier, float64) {
	switch operator {
	case 1:
		return carrierMobile, mobileQPS
	case 2:
		return carrierTelecom, telecomQPS
	default:
		return carrierGeneric, 0
	}
}

func newOwner() string {
	var bytes [12]byte
	if _, err := rand.Read(bytes[:]); err == nil {
		return "consumer-" + hex.EncodeToString(bytes[:])
	}
	return fmt.Sprintf("consumer-%d", time.Now().UnixNano())
}