// Package monthlyusagehistory produces the previous-month usage history run.
//
// It writes the TDengine monthly table, a different target from
// producer/monthlyusage (which fills the MySQL current-month snapshot). Apart
// from the period and the destination it is structurally identical: scan the
// eligible cards once, group by channel, and enqueue one Redis ZSet task per
// card. The zset consumer owns pacing, retries, failure pooling, archiving and
// replay.
//
// The run period is a month (yyyyMM) and travels in zset.Task.QueryMonth, which
// is a period identifier rather than a month name; see that field's
// documentation.
package monthlyusagehistory

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"strings"
	"sync/atomic"
	"time"

	"github.com/gogf/gf/v2/database/gdb"
	"github.com/gogf/gf/v2/frame/g"

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

// FullSyncOperation identifies a monthly usage history run in run metadata,
// archives and replay. contract.SyncKindForOperation maps it to
// traffic_usage_monthly.
const FullSyncOperation = model.OperationPreviousMonthUsageFullSync

// scopeSampleLimit bounds how many channel names are copied into run metadata.
// The run hash and its archive JSON must stay bounded regardless of how many
// channels a scope selects.
const scopeSampleLimit = 10

// scopeDigest is a stable short fingerprint of the selected channels. It lets
// operators confirm two runs covered the same scope without storing the list.
func scopeDigest(scope *model.SyncScope) string {
	if scope == nil || len(scope.Channels) == 0 {
		return ""
	}
	digest := sha256.Sum256([]byte(strings.Join(scope.Channels, "\n")))
	return hex.EncodeToString(digest[:8])
}

type Config struct {
	ScanMode             string
	ScanConcurrency      int
	PageSize             int
	GlobalHighWaterMark  int64
	ChannelHighWaterMark int64
	PollInterval         time.Duration
	RunLockKey           string
	RunLockTTL           time.Duration
	RoundRetryEnabled    bool
	MaxRounds            int
	FailureRetention     time.Duration
}

func (c Config) withDefaults() Config {
	if c.ScanMode == "" {
		c.ScanMode = "serial"
	}
	if c.ScanConcurrency == 0 {
		c.ScanConcurrency = 4
	}
	if c.PageSize <= 0 {
		c.PageSize = 100
	}
	if c.PollInterval <= 0 {
		c.PollInterval = 2 * time.Second
	}
	if c.RunLockKey == "" {
		c.RunLockKey = "monthly-history:producer:active"
	}
	if c.RunLockTTL <= 0 {
		c.RunLockTTL = 5 * time.Minute
	}
	if c.MaxRounds <= 0 {
		c.MaxRounds = 3
	}
	if c.FailureRetention <= 0 {
		c.FailureRetention = 30 * 24 * time.Hour
	}
	return c
}

type CardScanner interface {
	SnapshotMaxID(context.Context) (int64, error)
	Scan(context.Context, int64, int64, int) ([]Card, error)
}

type Card struct {
	ID          int64
	ICCID       string
	MSISDN      string
	IMSI        string
	ChanKey     string
	TempKey     string
	PackageCode string
	SyncEnabled bool
	Operator    int
}

type channelMetadata struct {
	ChanKey     string `orm:"chan_key"`
	TempKey     string `orm:"temp_key"`
	SyncEnabled bool   `orm:"sync_enabled"`
	Operator    int    `orm:"operator"`
}

type DatabaseScanner struct {
	channelMetadata map[string]channelMetadata
}

func (s *DatabaseScanner) SnapshotMaxID(ctx context.Context) (int64, error) {
	var card entity.TSimCardInfo
	cardTable := dao.TSimCardInfo.Table()
	columns := dao.TSimCardInfo.Columns()
	if err := monthlyHistoryEligibleCardModel(ctx).
		Fields(cardTable + "." + columns.Id + " AS " + columns.Id).
		OrderDesc(cardTable + "." + columns.Id).
		Limit(1).
		Scan(&card); err != nil {
		return 0, fmt.Errorf("read card snapshot max id: %w", err)
	}
	return card.Id, nil
}

func (s *DatabaseScanner) Scan(ctx context.Context, lastID, maxID int64, pageSize int) ([]Card, error) {
	if maxID <= lastID {
		return nil, nil
	}
	if lastID == 0 || s.channelMetadata == nil {
		var metadata []channelMetadata
		if err := g.DB().Model("t_sim_channels").Fields("chan_key,temp_key,sync_enabled,operator").Scan(&metadata); err != nil {
			return nil, fmt.Errorf("scan channel metadata: %w", err)
		}
		s.channelMetadata = make(map[string]channelMetadata, len(metadata))
		for _, item := range metadata {
			s.channelMetadata[item.ChanKey] = item
		}
	}
	var rows []entity.TSimCardInfo
	columns := dao.TSimCardInfo.Columns()
	cardTable := dao.TSimCardInfo.Table()
	query := monthlyHistoryEligibleCardModel(ctx).
		Fields(
			cardTable+"."+columns.Id+" AS "+columns.Id,
			cardTable+"."+columns.Iccid+" AS "+columns.Iccid,
			cardTable+"."+columns.Msisdn+" AS "+columns.Msisdn,
			cardTable+"."+columns.Imsi+" AS "+columns.Imsi,
			cardTable+"."+columns.ChanKey+" AS "+columns.ChanKey,
			// The monthly usage child table stores the package code and the
			// diagnosis page reads it back, so it must be carried on the task.
			cardTable+"."+columns.CurrentPackageCode+" AS "+columns.CurrentPackageCode,
		).
		WhereGT(cardTable+"."+columns.Id, lastID).
		WhereLTE(cardTable+"."+columns.Id, maxID).
		OrderAsc(cardTable + "." + columns.Id).
		Limit(pageSize)
	if err := query.Scan(&rows); err != nil {
		return nil, fmt.Errorf("scan card information: %w", err)
	}
	cards := make([]Card, 0, len(rows))
	for _, row := range rows {
		channel := s.channelMetadata[row.ChanKey]
		cards = append(cards, Card{
			ID:          row.Id,
			ICCID:       row.Iccid,
			MSISDN:      row.Msisdn,
			IMSI:        row.Imsi,
			ChanKey:     row.ChanKey,
			TempKey:     channel.TempKey,
			PackageCode: row.CurrentPackageCode,
			SyncEnabled: channel.SyncEnabled,
			Operator:    channel.Operator,
		})
	}
	return cards, nil
}

// monthlyHistoryEligibleCardModel mirrors the monthly usage eligibility rule and
// the legacy admin SyncMonthly scan: the card must belong to a live department,
// carry an ICCID and channel, and have a current package code.
func monthlyHistoryEligibleCardModel(ctx context.Context) *gdb.Model {
	cardTable := dao.TSimCardInfo.Table()
	deptTable := dao.SysDept.Table()
	cardColumns := dao.TSimCardInfo.Columns()
	deptColumns := dao.SysDept.Columns()

	return dao.TSimCardInfo.Ctx(ctx).
		InnerJoin(deptTable, cardTable+"."+cardColumns.DeptId+" = "+deptTable+"."+deptColumns.DeptId+" AND "+cardTable+"."+cardColumns.TenantId+" = "+deptTable+"."+deptColumns.TenantId).
		Where(cardTable+"."+cardColumns.Iccid+" <> ?", "").
		Where(cardTable+"."+cardColumns.ChanKey+" <> ?", "").
		Where(cardTable+"."+cardColumns.CurrentPackageCode+" <> ?", "").
		WhereNull(deptTable + "." + deptColumns.DeletedAt)
}

type Producer struct {
	scanner     CardScanner
	store       *zset.Store
	config      Config
	running     atomic.Int32
	owner       string
	observation *supplyObservation
}

type productionStats struct {
	pages               int64
	scanned             int64
	eligible            int64
	enqueued            int64
	skippedEmptyChan    int64
	skippedNoTemplate   int64
	skippedSyncDisabled int64
	skippedOutOfScope   int64
	stoppedTasks        int64
	channels            map[string]struct{}
	scopeChannels       int
	skippedChannels     int
}

func (s *productionStats) channelCount() int {
	return len(s.channels)
}

func (s *productionStats) log(ctx context.Context, status, runID, queryMonth, stage, reason string, startedAt time.Time, maxID int64, err error) {
	message := fmt.Sprintf("event=monthly_usage_history_producer status=%s runId=%s queryMonth=%s snapshotMaxId=%d pages=%d scanned=%d eligible=%d enqueued=%d skippedEmptyChan=%d skippedNoTemplate=%d skippedSyncDisabled=%d skippedOutOfScope=%d stoppedTasks=%d channelCount=%d scopeChannels=%d skippedChannels=%d durationMs=%d", status, runID, queryMonth, maxID, s.pages, s.scanned, s.eligible, s.enqueued, s.skippedEmptyChan, s.skippedNoTemplate, s.skippedSyncDisabled, s.skippedOutOfScope, s.stoppedTasks, s.channelCount(), s.scopeChannels, s.skippedChannels, time.Since(startedAt).Milliseconds())
	if s.scanned > 0 {
		message += fmt.Sprintf(" cardsPerSec=%.2f", float64(s.scanned)/time.Since(startedAt).Seconds())
	}
	if stage != "" {
		message += " stage=" + stage
	}
	if reason != "" {
		message += " reason=" + reason
	}
	if err != nil {
		message += fmt.Sprintf(" error=%q", err.Error())
	}
	if status == "failed" {
		g.Log().Error(ctx, message)
		return
	}
	if status == "skipped" {
		g.Log().Warning(ctx, message)
		return
	}
	g.Log().Info(ctx, message)
}

func New(scanner CardScanner, store *zset.Store, config Config) (*Producer, error) {
	if scanner == nil || store == nil {
		return nil, fmt.Errorf("monthly usage history producer dependencies are incomplete")
	}
	config = config.withDefaults()
	if err := ValidateScanSettings(config.ScanMode, config.ScanConcurrency); err != nil {
		return nil, err
	}
	if config.ScanMode == "channel" {
		if _, ok := scanner.(channelScanner); !ok {
			return nil, fmt.Errorf("channel scan requires ListChannels and ScanChannel")
		}
	}
	return &Producer{scanner: scanner, store: store, config: config, owner: newOwner()}, nil
}

var ErrAlreadyRunning = fmt.Errorf("monthly usage history producer already running")

// ErrRunFinished is returned when a redelivered start message names a run that
// already reached a terminal state. The run must not be resurrected; callers
// should log and ack instead of retrying.
var ErrRunFinished = fmt.Errorf("monthly usage history run already finished")

// runAlreadyFinished reports whether runID exists and has already completed.
// A missing run is not an error: the message may be the first delivery.
func (p *Producer) runAlreadyFinished(ctx context.Context, runID string) (bool, error) {
	meta, err := p.store.GetRunMeta(ctx, runID)
	if err != nil {
		if errors.Is(err, zset.ErrRunNotFound) {
			return false, nil
		}
		return false, err
	}
	return meta.Status == "completed" || meta.Status == "completed_with_failures", nil
}

// Produce scans the eligible cards once and enqueues one task per card.
//
// message.QueryMonth (yyyyMM) is stored in zset.Task.QueryMonth, the queue's
// period-identifier field, so the zset store, run metadata, archive, replay and
// Lua scripts need no change for a monthly history run.
func (p *Producer) Produce(ctx context.Context, message model.TrafficUsageMonthlyHistoryStartMessage) (err error) {
	startedAt := time.Now()
	stats := productionStats{channels: make(map[string]struct{})}
	if !p.running.CompareAndSwap(0, 1) {
		stats.log(ctx, "skipped", message.RunID, message.QueryMonth, "precheck", "already_running", startedAt, 0, nil)
		return ErrAlreadyRunning
	}
	defer p.running.Store(0)

	if message.RunID == "" {
		err = fmt.Errorf("runId is empty")
		stats.log(ctx, "failed", message.RunID, message.QueryMonth, "validate", "", startedAt, 0, err)
		return err
	}
	if message.QueryMonth == "" {
		err = fmt.Errorf("queryMonth is empty")
		stats.log(ctx, "failed", message.RunID, message.QueryMonth, "validate", "", startedAt, 0, err)
		return err
	}
	scope, err := model.NormalizeScope(message.Scope)
	if err != nil {
		stats.log(ctx, "failed", message.RunID, message.QueryMonth, "validate", "invalid_scope", startedAt, 0, err)
		return err
	}
	stats.scopeChannels = scope.ChannelCount()
	// A redelivered start message must not resurrect a run that already
	// finished; that would create a second producer pass over the same RunID.
	if p.config.RoundRetryEnabled {
		if finished, checkErr := p.runAlreadyFinished(ctx, message.RunID); checkErr != nil {
			stats.log(ctx, "failed", message.RunID, message.QueryMonth, "run_state", "", startedAt, 0, checkErr)
			return checkErr
		} else if finished {
			stats.log(ctx, "skipped", message.RunID, message.QueryMonth, "run_state", "run_finished", startedAt, 0, nil)
			return ErrRunFinished
		}
	}
	acquired, err := p.store.AcquireLockOrTakeover(ctx, p.config.RunLockKey, p.owner, message.RunID, p.config.RunLockTTL)
	if err != nil {
		stats.log(ctx, "failed", message.RunID, message.QueryMonth, "lock", "", startedAt, 0, err)
		return err
	}
	if !acquired {
		stats.log(ctx, "skipped", message.RunID, message.QueryMonth, "lock", "already_running", startedAt, 0, nil)
		return ErrAlreadyRunning
	}
	runCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	defer func() {
		releaseCtx, releaseCancel := context.WithTimeout(context.Background(), time.Second)
		defer releaseCancel()
		_, _ = p.store.ReleaseLock(releaseCtx, p.config.RunLockKey, p.owner)
	}()
	pending, err := p.hasPendingTasks(ctx, message.RunID)
	if err != nil {
		stats.log(ctx, "failed", message.RunID, message.QueryMonth, "pending_check", "", startedAt, 0, err)
		return err
	}
	if pending {
		stats.log(ctx, "skipped", message.RunID, message.QueryMonth, "pending_check", "pending_tasks", startedAt, 0, nil)
		return ErrAlreadyRunning
	}
	renewErr := make(chan error, 1)
	go p.renewLock(runCtx, cancel, renewErr)
	ctx = runCtx
	if message.Operation == "" {
		message.Operation = FullSyncOperation
	}
	if p.config.RoundRetryEnabled {
		if err = p.store.CreateRun(ctx, zset.RunMeta{
			RunID:               message.RunID,
			Operation:           message.Operation,
			QueryMonth:          message.QueryMonth,
			Status:              "producing",
			CurrentRound:        1,
			ScopeChannels:       scope.ChannelCount(),
			ScopeDigest:         scopeDigest(scope),
			ScopeChannelsSample: strings.Join(scope.ChannelSample(scopeSampleLimit), ","),
		}); err != nil {
			stats.log(ctx, "failed", message.RunID, message.QueryMonth, "run_init", "", startedAt, 0, err)
			return err
		}
	}
	maxID, err := p.scanner.SnapshotMaxID(ctx)
	if err != nil {
		stats.log(ctx, "failed", message.RunID, message.QueryMonth, "snapshot", "", startedAt, 0, err)
		return err
	}
	g.Log().Infof(ctx, "event=monthly_usage_history_producer status=started runId=%s queryMonth=%s snapshotMaxId=%d scopeChannels=%d scanMode=%s scanConcurrency=%d pageSize=%d", message.RunID, message.QueryMonth, maxID, scope.ChannelCount(), p.config.ScanMode, p.config.ScanConcurrency, p.config.PageSize)
	stage := "scan"
	defer func() {
		if err != nil {
			stats.log(ctx, "failed", message.RunID, message.QueryMonth, stage, "", startedAt, maxID, err)
			return
		}
		stats.log(ctx, "completed", message.RunID, message.QueryMonth, "", "", startedAt, maxID, nil)
	}()
	if p.config.RoundRetryEnabled && p.config.ScanMode == "channel" {
		stage = "channel_scan"
		if err = p.produceChannels(ctx, message, scope, maxID, startedAt, &stats); err != nil {
			if isProductionClosed(err) {
				// A same-Run redelivery after producerDone leaves consumption alone.
				err = nil
				return nil
			}
			return err
		}
		// Verify ownership immediately before closing production. The periodic
		// renewer still cancels scanners and capacity waits on lease loss.
		var owned bool
		owned, err = p.store.RenewLock(ctx, p.config.RunLockKey, p.owner, p.config.RunLockTTL)
		if err != nil {
			return err
		}
		if !owned {
			return fmt.Errorf("producer lock token is no longer owned")
		}
		if err = ctx.Err(); err != nil {
			return err
		}
		stage = "producer_done"
		err = p.store.MarkProducerDone(ctx, message.RunID)
		return err
	}
	lastID := int64(0)
	for lastID < maxID {
		stage = "capacity"
		if err = p.waitForCapacity(ctx, message.RunID); err != nil {
			return err
		}
		stage = "scan"
		cards, scanErr := p.scanner.Scan(ctx, lastID, maxID, p.config.PageSize)
		if scanErr != nil {
			err = scanErr
			return err
		}
		if len(cards) == 0 {
			break
		}
		stats.pages++
		stats.scanned += int64(len(cards))
		groupedTasks := make(map[string][]zset.Task)
		for _, card := range cards {
			lastID = card.ID
			if card.ChanKey == "" {
				stats.skippedEmptyChan++
				g.Log().Warningf(ctx, "event=monthly_usage_history_card status=skipped stage=validate cardId=%d reason=empty_chan_key", card.ID)
				continue
			}
			// Scope is applied before the sync_enabled and template checks so a
			// partial run only ever reports on the channels it was asked for.
			if !scope.Allows(card.ChanKey) {
				stats.skippedOutOfScope++
				continue
			}
			if card.SyncEnabled {
				stats.skippedSyncDisabled++
				g.Log().Warningf(ctx, "event=monthly_usage_history_card status=skipped stage=validate cardId=%d chanKey=%s reason=sync_disabled", card.ID, card.ChanKey)
				continue
			}
			if card.TempKey == "" {
				stats.skippedNoTemplate++
				g.Log().Warningf(ctx, "event=monthly_usage_history_card status=skipped stage=validate cardId=%d chanKey=%s reason=missing_channel_template", card.ID, card.ChanKey)
				continue
			}
			stats.eligible++
			stats.channels[card.ChanKey] = struct{}{}
			task := zset.Task{
				TaskID:      zset.StableTaskID(message.RunID, message.Operation, card.ID),
				RunID:       message.RunID,
				Operation:   message.Operation,
				QueryMonth:  message.QueryMonth,
				ChanKey:     card.ChanKey,
				TempKey:     card.TempKey,
				Operator:    card.Operator,
				CardID:      card.ID,
				ICCID:       card.ICCID,
				MSISDN:      card.MSISDN,
				IMSI:        card.IMSI,
				PackageCode: card.PackageCode,
			}
			groupedTasks[card.ChanKey] = append(groupedTasks[card.ChanKey], task)
		}
		now := time.Now()
		for chanKey, tasks := range groupedTasks {
			stage = "enqueue"
			if err = p.waitForChannelCapacity(ctx, message.RunID, chanKey); err != nil {
				return err
			}
			var added int
			var enqueueErr error
			if p.config.RoundRetryEnabled {
				added, enqueueErr = p.store.EnqueueRunBatch(ctx, message.RunID, chanKey, tasks, now)
			} else {
				added, enqueueErr = p.store.EnqueueBatch(ctx, chanKey, tasks, now)
			}
			if enqueueErr != nil {
				err = enqueueErr
				return err
			}
			stats.enqueued += int64(added)
			if added < len(tasks) {
				stats.stoppedTasks += int64(len(tasks) - added)
			}
			if added == 0 {
				g.Log().Warningf(ctx, "event=monthly_usage_history_channel status=skipped runId=%s queryMonth=%s chanKey=%s count=%d reason=stopped", message.RunID, message.QueryMonth, chanKey, len(tasks))
			}
		}
	}
	if p.config.RoundRetryEnabled {
		if err = p.store.MarkProducerDone(ctx, message.RunID); err != nil {
			return err
		}
	}
	return nil
}

func (p *Producer) hasPendingTasks(ctx context.Context, currentRunID ...string) (bool, error) {
	currentID := ""
	if len(currentRunID) > 0 {
		currentID = currentRunID[0]
	}
	if p.config.RoundRetryEnabled {
		runs, err := p.store.ActiveRuns(ctx)
		if err != nil {
			return false, err
		}
		for _, runID := range runs {
			if runID == currentID {
				// A redelivered message for the same run may resume a producer
				// that stopped before producerDone. Stable task IDs make the
				// rescan idempotent; other runs still provide backpressure.
				continue
			}
			meta, metaErr := p.store.GetRunMeta(ctx, runID)
			if metaErr != nil {
				// The active index is a global key written outside the run-scoped
				// Lua transaction, so it can list a run whose hash is already gone
				// (expired, evicted, or cleaned by hand). Returning an error here
				// would surface as a non-retryable failure and make the consumer
				// redeliver this request forever. Treat it as a stale index entry
				// and drop it.
				if errors.Is(metaErr, zset.ErrRunNotFound) {
					if _, removeErr := p.store.RedisClient().Do(ctx, "SREM", p.store.ActiveRunsKey(), runID); removeErr != nil {
						return false, removeErr
					}
					g.Log().Warningf(ctx, "event=monthly_usage_history_producer status=skipped stage=pending_check reason=stale_index runId=%s", runID)
					continue
				}
				return false, metaErr
			}
			if meta.Status != "producing" && meta.Status != "running" {
				continue
			}
			// Channel blocking can set running before production ends. Only
			// producerDone identifies consumer-owned work; replay has its own lock.
			if meta.ProducerDone || meta.SyncType == "replay" {
				return true, nil
			}
			// Recheck the unique producer lease immediately before reclaiming
			// another run. A lost lease or an unspecified current run is not proof
			// that the old producer has stopped.
			if currentID == "" {
				return true, nil
			}
			owned, lockErr := p.store.RenewLock(ctx, p.config.RunLockKey, p.owner, p.config.RunLockTTL)
			if lockErr != nil {
				return false, lockErr
			}
			if !owned {
				return true, nil
			}
			// Close production, not the run: consumers retain queued, leased and
			// next-round work and apply retention/index cleanup on normal completion.
			reclaimed, reclaimErr := p.store.MarkProducerAbandoned(ctx, runID)
			if reclaimErr != nil {
				return false, reclaimErr
			}
			g.Log().Warningf(ctx, "event=monthly_usage_history_producer status=skipped stage=pending_check reason=zombie_run runId=%s reclaimed=%t activeCount=%d inflightCount=%d nextRoundCount=%d", runID, reclaimed, meta.ActiveCount, meta.InflightCount, meta.NextRoundCount)
			if !reclaimed || meta.ActiveCount > 0 || meta.InflightCount > 0 || meta.NextRoundCount > 0 {
				return true, nil
			}
		}
	}
	channels, err := p.store.ActiveChannels(ctx)
	if err != nil {
		return false, err
	}
	for _, chanKey := range channels {
		size, sizeErr := p.store.Size(ctx, chanKey)
		if sizeErr != nil {
			return false, sizeErr
		}
		if size > 0 {
			return true, nil
		}
	}
	return false, nil
}

func (p *Producer) renewLock(ctx context.Context, cancel context.CancelFunc, errors chan<- error) {
	interval := p.config.RunLockTTL / 3
	if interval <= 0 {
		interval = time.Second
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			ok, err := p.store.RenewLock(ctx, p.config.RunLockKey, p.owner, p.config.RunLockTTL)
			if err != nil || !ok {
				if err == nil {
					err = fmt.Errorf("producer lock token is no longer owned")
				}
				select {
				case errors <- fmt.Errorf("renew producer lock: %w", err):
				default:
				}
				cancel()
				return
			}
		}
	}
}

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

func (p *Producer) waitForCapacity(ctx context.Context, runID ...string) error {
	if p.config.GlobalHighWaterMark <= 0 {
		return nil
	}
	for {
		if p.config.RoundRetryEnabled && len(runID) > 0 && runID[0] != "" {
			meta, err := p.store.GetRunMeta(ctx, runID[0])
			if err != nil {
				return err
			}
			channels, err := p.store.RunChannels(ctx, runID[0])
			if err != nil {
				return err
			}
			var total int64
			for _, chanKey := range channels {
				key, keyErr := p.store.RoundActiveKey(runID[0], meta.CurrentRound, chanKey)
				if keyErr != nil {
					return keyErr
				}
				size, sizeErr := p.store.SizeKey(ctx, key)
				if sizeErr != nil {
					return sizeErr
				}
				total += size
			}
			if total < p.config.GlobalHighWaterMark {
				return nil
			}
			if err = sleepContext(ctx, p.config.PollInterval); err != nil {
				return err
			}
			continue
		}
		channels, err := p.store.ActiveChannels(ctx)
		if err != nil {
			return err
		}
		var total int64
		for _, chanKey := range channels {
			size, sizeErr := p.store.Size(ctx, chanKey)
			if sizeErr != nil {
				return sizeErr
			}
			total += size
		}
		if total < p.config.GlobalHighWaterMark {
			return nil
		}
		if err = sleepContext(ctx, p.config.PollInterval); err != nil {
			return err
		}
	}
}

func (p *Producer) waitForChannelCapacity(ctx context.Context, runIDAndChan ...string) error {
	if p.config.ChannelHighWaterMark <= 0 {
		return nil
	}
	chanKey := ""
	runID := ""
	if len(runIDAndChan) == 1 {
		chanKey = runIDAndChan[0]
	} else if len(runIDAndChan) >= 2 {
		runID, chanKey = runIDAndChan[0], runIDAndChan[1]
	}
	for {
		var size int64
		var err error
		if p.config.RoundRetryEnabled && runID != "" {
			meta, metaErr := p.store.GetRunMeta(ctx, runID)
			if metaErr != nil {
				return metaErr
			}
			key, keyErr := p.store.RoundActiveKey(runID, meta.CurrentRound, chanKey)
			if keyErr != nil {
				return keyErr
			}
			size, err = p.store.SizeKey(ctx, key)
		} else {
			size, err = p.store.Size(ctx, chanKey)
		}
		if err != nil {
			return err
		}
		if size < p.config.ChannelHighWaterMark {
			return nil
		}
		if err = sleepContext(ctx, p.config.PollInterval); err != nil {
			return err
		}
	}
}

func sleepContext(ctx context.Context, duration time.Duration) error {
	timer := time.NewTimer(duration)
	defer timer.Stop()
	select {
	case <-timer.C:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}