iotlcm/worker-svr/internal/queue/zset/run.go

package zset

// This file contains the run-scoped queue protocol used by the full traffic
// usage synchronizer.  The original channel-scoped API in store.go remains
// intentionally unchanged; run keys are separate so old, in-flight work can
// continue to be consumed while new producers use round barriers.

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/gogf/gf/v2/errors/gerror"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/supply"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
)

// ErrRunNotFound reports that a run's metadata hash does not exist. Callers use
// errors.Is so a stale active-index entry can be told apart from a real read
// failure.
var ErrRunNotFound = errors.New("run not found")

// RunMeta is the durable state for one full-sync run. Counts are maintained by
// the Lua transitions below rather than inferred from logs.
type RunMeta struct {
	SupplyObservation    *supply.Snapshot `json:"supplyObservation,omitempty"`
	RunID                string           `json:"runId"`
	Operation            string           `json:"operation"`
	QueryMonth           string           `json:"queryMonth"`
	Status               string           `json:"status"`
	ProducerDone         bool             `json:"producerDone"`
	CurrentRound         int              `json:"currentRound"`
	ActiveCount          int64            `json:"activeCount"`
	InflightCount        int64            `json:"inflightCount"`
	NextRoundCount       int64            `json:"nextRoundCount"`
	SuccessCount         int64            `json:"successCount"`
	FailedCount          int64            `json:"failedCount"`
	CreatedAt            time.Time        `json:"createdAt"`
	ProducerCompletedAt  time.Time        `json:"producerCompletedAt"`
	CompletedAt          time.Time        `json:"completedAt"`
	SyncType             string           `json:"syncType,omitempty"`
	SourceRunID          string           `json:"sourceRunId,omitempty"`
	ReplayCycle          int              `json:"replayCycle,omitempty"`
	TriggerSource        string           `json:"triggerSource,omitempty"`
	ScannedCount         int64            `json:"scannedCount,omitempty"`
	EnqueuedCount        int64            `json:"enqueuedCount,omitempty"`
	UpstreamAttemptCount int64            `json:"upstreamAttemptCount,omitempty"`
	BatchCount           int64            `json:"batchCount,omitempty"`
	MaxRound             int              `json:"maxRound,omitempty"`
	PendingFailureCount  int64            `json:"pendingFailureCount,omitempty"`
	StartedAt            *time.Time       `json:"startedAt,omitempty"`
	ScanCompletedAt      *time.Time       `json:"scanCompletedAt,omitempty"`
	ArchiveStatus        string           `json:"archiveStatus,omitempty"`
	ArchiveVersion       int              `json:"archiveVersion,omitempty"`
	ArchivePath          string           `json:"archivePath,omitempty"`
	ArchiveChecksum      string           `json:"archiveChecksum,omitempty"`
	ArchivedAt           *time.Time       `json:"archivedAt,omitempty"`
	ReplayDisposition    string           `json:"replayDisposition,omitempty"`
	// Scope fields describe an optional channel whitelist. ScopeChannels == 0
	// means the run covered every eligible channel (a full sync), which is also
	// what legacy runs without these fields report.
	ScopeChannels       int          `json:"scopeChannels,omitempty"`
	ScopeDigest         string       `json:"scopeDigest,omitempty"`
	ScopeChannelsSample string       `json:"scopeChannelsSample,omitempty"`
	RoundStats          []RoundStats `json:"rounds,omitempty"`
}

// RoundStats is the bounded aggregate persisted with a run. It deliberately
// contains no task identity or error detail so the Run hash stays small.
type RoundStats struct {
	RunID                string     `json:"runId"`
	Round                int        `json:"round"`
	EnqueuedCount        int64      `json:"enqueuedCount"`
	ActiveCount          int64      `json:"activeCount"`
	NextRoundCount       int64      `json:"nextRoundCount"`
	SuccessCount         int64      `json:"successCount"`
	FailedCount          int64      `json:"failedCount"`
	UpstreamAttemptCount int64      `json:"upstreamAttemptCount"`
	BatchCount           int64      `json:"batchCount"`
	StartedAt            *time.Time `json:"startedAt,omitempty"`
	CompletedAt          *time.Time `json:"completedAt,omitempty"`
}

// TaskState tracks attempts independently from the stable ZSet member.
type TaskState struct {
	Round       int       `json:"round"`
	Status      string    `json:"status"`
	Attempts    int       `json:"attempts"`
	LastCode    int       `json:"lastCode,omitempty"`
	LastErrCode string    `json:"lastErrCode,omitempty"`
	LastMessage string    `json:"lastMessage,omitempty"`
	UpdatedAt   time.Time `json:"updatedAt"`
}

// FailureRecord is persisted for permanent failures and exhausted busy
// retries. It deliberately contains the task identity and response fields so
// an external reviewer can diagnose a run without the original queue member.
type FailureRecord struct {
	Version       int    `json:"version"`
	TaskID        string `json:"taskId"`
	OriginalRunID string `json:"originalRunId"`
	LatestRunID   string `json:"latestRunId"`
	QueryMonth    string `json:"queryMonth"`
	Operation     string `json:"operation"`
	CardID        int64  `json:"cardId"`
	ICCID         string `json:"iccid"`
	MSISDN        string `json:"msisdn"`
	IMSI          string `json:"imsi"`
	ChanKey       string `json:"chanKey"`
	TempKey       string `json:"tempKey"`
	// PackageCode is additive; older failure records decode it as empty. It is
	// preserved so a replayed daily-usage card can restore the original package
	// code instead of writing an empty value into the TDengine child table.
	PackageCode   string    `json:"packageCode,omitempty"`
	Carrier       string    `json:"carrier"`
	FailureType   string    `json:"failureType"`
	Round         int       `json:"round"`
	Attempts      int       `json:"attempts"`
	Code          int       `json:"code"`
	ErrCode       string    `json:"errCode"`
	Message       string    `json:"message"`
	ReviewStatus  string    `json:"reviewStatus"`
	ReviewRemark  string    `json:"reviewRemark"`
	ReviewedBy    string    `json:"reviewedBy"`
	ReviewedAt    string    `json:"reviewedAt"`
	RetryCycle    int       `json:"retryCycle"`
	ReplayRunID   string    `json:"replayRunId"`
	FirstFailedAt time.Time `json:"firstFailedAt"`
	LastFailedAt  time.Time `json:"lastFailedAt"`
}

// TaskOutcome is supplied by a consumer when atomically finalizing a lease.
// ErrorClass is intentionally generic: the consumer may classify protocol
// responses without making the queue package depend on service packages.
type TaskOutcome struct {
	Success      bool
	Retry        bool
	Retryable    bool
	ErrorClass   string
	FailureType  string
	Code         int
	ErrCode      string
	Message      string
	BlockChannel bool
	// Covered marks a terminal outcome that answered definitively but carries
	// no value ("the carrier has no record of this card", "no IMEI reported
	// yet"). It is not a failure: the task leaves the queue like any other
	// final outcome and counts as completed work, but it must never be written
	// to the failure pool or counted in failedCount. The zero value preserves
	// the historical behaviour, so a caller that does not classify the case is
	// unaffected.
	Covered bool
	Failure *FailureRecord
}

// taskStatusCovered is the task-state value persisted for a covered outcome.
// It is distinct from succeeded/failed because the failure-review page keys off
// the persisted state: a card with no carrier record did not fail, it simply
// has no value to show.
const taskStatusCovered = "covered"

// Run key builders. Every key contains the same Redis hash-tag ({runId}), so
// multi-key Lua transitions remain valid when Redis Cluster is enabled.
func (s *Store) runTag(runID string) (string, error) {
	if strings.TrimSpace(runID) == "" {
		return "", fmt.Errorf("runId is empty")
	}
	return s.prefix + "{" + runID + "}", nil
}

func (s *Store) RunMetaKey(runID string) (string, error) {
	tag, err := s.runTag(runID)
	if err != nil {
		return "", err
	}
	return tag + ":meta", nil
}
func (s *Store) RunChannelsKey(runID string) (string, error) {
	tag, err := s.runTag(runID)
	if err != nil {
		return "", err
	}
	return tag + ":channels", nil
}
func (s *Store) RoundActiveKey(runID string, round int, chanKey string) (string, error) {
	tag, err := s.runTag(runID)
	if err != nil {
		return "", err
	}
	if round <= 0 {
		return "", fmt.Errorf("round must be positive")
	}
	if strings.TrimSpace(chanKey) == "" {
		return "", fmt.Errorf("chanKey is empty")
	}
	return fmt.Sprintf("%s:round:%d:active:%s", tag, round, chanKey), nil
}
func (s *Store) RoundWaitingKey(runID string, round int, chanKey string) (string, error) {
	tag, err := s.runTag(runID)
	if err != nil {
		return "", err
	}
	if round <= 0 {
		return "", fmt.Errorf("round must be positive")
	}
	if strings.TrimSpace(chanKey) == "" {
		return "", fmt.Errorf("chanKey is empty")
	}
	return fmt.Sprintf("%s:round:%d:waiting:%s", tag, round, chanKey), nil
}
func (s *Store) RoundLeaseKey(runID string, round int, chanKey string) (string, error) {
	tag, err := s.runTag(runID)
	if err != nil {
		return "", err
	}
	if round <= 0 {
		return "", fmt.Errorf("round must be positive")
	}
	if strings.TrimSpace(chanKey) == "" {
		return "", fmt.Errorf("chanKey is empty")
	}
	return fmt.Sprintf("%s:round:%d:lease:%s", tag, round, chanKey), nil
}
func (s *Store) TaskStateKey(runID string) (string, error) {
	tag, err := s.runTag(runID)
	if err != nil {
		return "", err
	}
	return tag + ":task-state", nil
}
func (s *Store) FailedKey(runID string) (string, error) {
	tag, err := s.runTag(runID)
	if err != nil {
		return "", err
	}
	return tag + ":failed", nil
}
func (s *Store) FailedRunsKey() string { return s.prefix + "failed:runs" }
func (s *Store) ActiveRunsKey() string { return s.prefix + "runs:active" }

// Lower-case aliases keep key construction consistent with the legacy store
// helpers (leaseKey/blockedKey) and are useful to package-local tests.
func (s *Store) runMetaKey(runID string) (string, error)     { return s.RunMetaKey(runID) }
func (s *Store) runChannelsKey(runID string) (string, error) { return s.RunChannelsKey(runID) }
func (s *Store) roundActiveKey(runID string, round int, chanKey string) (string, error) {
	return s.RoundActiveKey(runID, round, chanKey)
}
func (s *Store) roundWaitingKey(runID string, round int, chanKey string) (string, error) {
	return s.RoundWaitingKey(runID, round, chanKey)
}
func (s *Store) roundLeaseKey(runID string, round int, chanKey string) (string, error) {
	return s.RoundLeaseKey(runID, round, chanKey)
}
func (s *Store) taskStateKey(runID string) (string, error) { return s.TaskStateKey(runID) }
func (s *Store) failedKey(runID string) (string, error)    { return s.FailedKey(runID) }

// CreateRun initializes metadata and the run channel set. Replaying the same
// run is idempotent: an existing metadata hash is left intact.
func (s *Store) CreateRun(ctx context.Context, meta RunMeta) error {
	if strings.TrimSpace(meta.RunID) == "" {
		return fmt.Errorf("runId is empty")
	}
	if meta.CurrentRound <= 0 {
		meta.CurrentRound = 1
	}
	if meta.Status == "" {
		meta.Status = "producing"
	}
	if meta.CreatedAt.IsZero() {
		meta.CreatedAt = time.Now()
	}
	metaKey, err := s.RunMetaKey(meta.RunID)
	if err != nil {
		return err
	}
	args := []any{meta.RunID, meta.Operation, meta.QueryMonth, meta.Status, boolString(meta.ProducerDone), meta.CurrentRound, meta.ActiveCount, meta.InflightCount, meta.NextRoundCount, meta.SuccessCount, meta.FailedCount, meta.CreatedAt.UTC().Format(time.RFC3339Nano)}
	created, err := s.redis.Eval(ctx, createRunScript, 1, []string{metaKey}, args)
	if err != nil {
		return fmt.Errorf("create run: %w", err)
	}
	if created.Int() == 0 {
		return fmt.Errorf("create run %q failed", meta.RunID)
	}
	if created.Int() == 2 {
		// Do not resurrect a completed run in the active index when an old
		// producer delivery is replayed, and reject accidental RunID reuse.
		stored, getErr := s.GetRunMeta(ctx, meta.RunID)
		if getErr != nil {
			return getErr
		}
		if (meta.Operation != "" && stored.Operation != "" && stored.Operation != meta.Operation) || (meta.QueryMonth != "" && stored.QueryMonth != "" && stored.QueryMonth != meta.QueryMonth) {
			return fmt.Errorf("run %q metadata does not match existing run", meta.RunID)
		}
		if stored.Status == "completed" || stored.Status == "completed_with_failures" {
			return nil
		}
	}
	// The active-run index is intentionally updated outside the run-key Lua
	// script. It is a global key and therefore lives in a different Redis
	// Cluster slot from flow:{runId}:* keys.
	if _, err = s.redis.Do(ctx, "SADD", s.ActiveRunsKey(), meta.RunID); err != nil {
		return fmt.Errorf("index active run: %w", err)
	}
	if err := s.setRunMetaExtensions(ctx, meta); err != nil {
		return err
	}
	return nil
}

func (s *Store) setRunMetaExtensions(ctx context.Context, meta RunMeta) error {
	args := []any{meta.RunID}
	add := func(name, value string) {
		if value != "" {
			args = append(args, name, value)
		}
	}
	addInt := func(name string, value int64) {
		if value != 0 {
			args = append(args, name, strconv.FormatInt(value, 10))
		}
	}
	addIntN := func(name string, value int) {
		if value != 0 {
			args = append(args, name, strconv.Itoa(value))
		}
	}
	addTime := func(name string, value *time.Time) {
		if value != nil && !value.IsZero() {
			add(name, value.UTC().Format(time.RFC3339Nano))
		}
	}
	add("syncType", meta.SyncType)
	add("sourceRunId", meta.SourceRunID)
	addIntN("replayCycle", meta.ReplayCycle)
	add("triggerSource", meta.TriggerSource)
	addInt("scannedCount", meta.ScannedCount)
	addInt("enqueuedCount", meta.EnqueuedCount)
	addInt("upstreamAttemptCount", meta.UpstreamAttemptCount)
	addInt("batchCount", meta.BatchCount)
	addIntN("maxRound", meta.MaxRound)
	addInt("pendingFailureCount", meta.PendingFailureCount)
	addTime("startedAt", meta.StartedAt)
	addTime("scanCompletedAt", meta.ScanCompletedAt)
	add("archiveStatus", meta.ArchiveStatus)
	addIntN("archiveVersion", meta.ArchiveVersion)
	add("archivePath", meta.ArchivePath)
	add("archiveChecksum", meta.ArchiveChecksum)
	addTime("archivedAt", meta.ArchivedAt)
	add("replayDisposition", meta.ReplayDisposition)
	addIntN("scopeChannels", meta.ScopeChannels)
	add("scopeDigest", meta.ScopeDigest)
	add("scopeChannelsSample", meta.ScopeChannelsSample)
	if len(args) == 1 {
		return nil
	}
	hsetArgs := append([]any{s.mustRunMetaKey(meta.RunID)}, args[1:]...)
	if _, err := s.redis.Do(ctx, "HSET", hsetArgs...); err != nil {
		return fmt.Errorf("set run metadata extensions: %w", err)
	}
	return nil
}

func (s *Store) mustRunMetaKey(runID string) string { key, _ := s.RunMetaKey(runID); return key }

// EnqueueRunBatch writes tasks to the active ZSet for the run's current round.
// The task ID is used as a hash field to make retries/idempotent producer
// deliveries no-ops. It also initializes task-state to queued.
func (s *Store) EnqueueRunBatch(ctx context.Context, runID, chanKey string, tasks []Task, nextExecuteAt time.Time) (int, error) {
	if len(tasks) == 0 {
		return 0, nil
	}
	if strings.TrimSpace(runID) == "" {
		return 0, fmt.Errorf("runId is empty")
	}
	if strings.TrimSpace(chanKey) == "" {
		return 0, fmt.Errorf("chanKey is empty")
	}
	if nextExecuteAt.IsZero() {
		nextExecuteAt = time.Now()
	}
	metaKey, err := s.RunMetaKey(runID)
	if err != nil {
		return 0, err
	}
	channelsKey, _ := s.RunChannelsKey(runID)
	stateKey, _ := s.TaskStateKey(runID)
	activeKey, _ := s.RoundActiveKey(runID, 1, chanKey) // script resolves round from metadata; key is replaced below
	// The active key must be for currentRound. Read metadata before writing; the
	// producer normally enqueues only during round one, while this also supports
	// replay producers that initialize another round.
	meta, getErr := s.GetRunMeta(ctx, runID)
	if getErr != nil {
		return 0, getErr
	}
	activeKey, _ = s.RoundActiveKey(runID, meta.CurrentRound, chanKey)
	args := make([]any, 0, len(tasks)*2+4)
	args = append(args, chanKey, nextExecuteAt.UnixMilli(), meta.CurrentRound, time.Now().UTC().Format(time.RFC3339Nano))
	for _, task := range tasks {
		if task.TaskID == "" {
			return 0, fmt.Errorf("taskId is empty")
		}
		if task.RunID != runID {
			return 0, fmt.Errorf("task runId %q does not match runId %q", task.RunID, runID)
		}
		if task.ChanKey != chanKey {
			return 0, fmt.Errorf("task chanKey %q does not match batch chanKey %q", task.ChanKey, chanKey)
		}
		task.NextExecuteAt = nextExecuteAt.UnixMilli()
		member, marshalErr := json.Marshal(task)
		if marshalErr != nil {
			return 0, fmt.Errorf("marshal task: %w", marshalErr)
		}
		args = append(args, task.TaskID, string(member))
	}
	result, err := s.redis.Eval(ctx, enqueueRunScript, 4, []string{metaKey, channelsKey, activeKey, stateKey}, args)
	if err != nil {
		return 0, fmt.Errorf("enqueue run batch: %w", err)
	}
	return result.Int(), nil
}

func (s *Store) MarkProducerDone(ctx context.Context, runID string) error {
	metaKey, err := s.RunMetaKey(runID)
	if err != nil {
		return err
	}
	result, err := s.redis.Eval(ctx, markProducerDoneScript, 1, []string{metaKey}, []any{time.Now().UTC().Format(time.RFC3339Nano)})
	if err != nil {
		return fmt.Errorf("mark producer done: %w", err)
	}
	if result.Int() == 0 {
		return fmt.Errorf("run %q not found", runID)
	}
	return nil
}

// MarkProducerAbandoned promotes a run whose producer died before
// MarkProducerDone so the normal lifecycle can finish it.
//
// The caller must hold the unique full-sync producer lock and exclude its own
// RunID. A run left in "producing" or "running" with producerDone=0 cannot close:
// completeRunIfIdle and advanceRunRound both require producerDone=1. It would
// therefore stay in the active index forever, its keys would never receive the
// retention TTL, and the consumer would poll it on every round. Setting
// producerDone=1 lets the consumer drain whatever was already enqueued and then
// close the run normally, which applies the retention TTL and releases the
// index.
//
// It refuses to touch replay runs: a replay target is created as "producing"
// too, but it is driven by the replay engine under its own lock, so its
// producerDone is not this producer's to set.
//
// Returns true when the run was actually promoted.
func (s *Store) MarkProducerAbandoned(ctx context.Context, runID string) (bool, error) {
	metaKey, err := s.RunMetaKey(runID)
	if err != nil {
		return false, err
	}
	result, err := s.redis.Eval(ctx, markProducerAbandonedScript, 1, []string{metaKey}, []any{time.Now().UTC().Format(time.RFC3339Nano)})
	if err != nil {
		return false, fmt.Errorf("mark producer abandoned: %w", err)
	}
	return result.Int() == 1, nil
}

func (s *Store) GetRunMeta(ctx context.Context, runID string) (RunMeta, error) {
	key, err := s.RunMetaKey(runID)
	if err != nil {
		return RunMeta{}, err
	}
	result, err := s.redis.Do(ctx, "HGETALL", key)
	if err != nil {
		return RunMeta{}, fmt.Errorf("get run metadata: %w", err)
	}
	values := result.MapStrStr()
	if len(values) == 0 {
		return RunMeta{}, fmt.Errorf("%w: %s", ErrRunNotFound, runID)
	}
	meta := RunMeta{RunID: runID, Operation: values["operation"], QueryMonth: values["queryMonth"], Status: values["status"], ProducerDone: parseBool(values["producerDone"]), CurrentRound: parseInt(values["currentRound"]), ActiveCount: parseInt64(values["activeCount"]), InflightCount: parseInt64(values["inflightCount"]), NextRoundCount: parseInt64(values["nextRoundCount"]), SuccessCount: parseInt64(values["successCount"]), FailedCount: parseInt64(values["failedCount"])}
	meta.CreatedAt = parseTime(values["createdAt"])
	meta.ProducerCompletedAt = parseTime(values["producerCompletedAt"])
	meta.CompletedAt = parseTime(values["completedAt"])
	meta.SyncType = values["syncType"]
	meta.SourceRunID = values["sourceRunId"]
	meta.ReplayCycle = parseInt(values["replayCycle"])
	meta.TriggerSource = values["triggerSource"]
	meta.ScannedCount = parseInt64(values["scannedCount"])
	meta.EnqueuedCount = parseInt64(values["enqueuedCount"])
	meta.UpstreamAttemptCount = parseInt64(values["upstreamAttemptCount"])
	meta.BatchCount = parseInt64(values["batchCount"])
	meta.MaxRound = parseInt(values["maxRound"])
	meta.PendingFailureCount = parseInt64(values["pendingFailureCount"])
	meta.StartedAt = parseOptionalTime(values["startedAt"])
	meta.ScanCompletedAt = parseOptionalTime(values["scanCompletedAt"])
	meta.ArchiveStatus = values["archiveStatus"]
	meta.ArchiveVersion = parseInt(values["archiveVersion"])
	meta.ArchivePath = values["archivePath"]
	meta.ArchiveChecksum = values["archiveChecksum"]
	meta.ArchivedAt = parseOptionalTime(values["archivedAt"])
	meta.ReplayDisposition = values["replayDisposition"]
	if raw := values["supplyObservation"]; raw != "" {
		var observation supply.Snapshot
		if err := json.Unmarshal([]byte(raw), &observation); err != nil || !observation.Valid() {
			observation = supply.Snapshot{SchemaVersion: 0, ResolutionMS: 1000, Partial: true}
		}
		meta.SupplyObservation = &observation
	}
	meta.ScopeChannels = parseInt(values["scopeChannels"])
	meta.ScopeDigest = values["scopeDigest"]
	meta.ScopeChannelsSample = values["scopeChannelsSample"]
	if raw := values["roundStats"]; raw != "" {
		if err := json.Unmarshal([]byte(raw), &meta.RoundStats); err != nil {
			return RunMeta{}, fmt.Errorf("decode run round stats: %w", err)
		}
	}
	return meta, nil
}

// SaveRoundStats replaces the bounded round summary JSON stored alongside the
// run metadata. The single run consumer is the writer, so one HSET is enough
// and avoids introducing another Redis key or coordination protocol.
func (s *Store) SaveRoundStats(ctx context.Context, runID string, stats []RoundStats) error {
	key, err := s.RunMetaKey(runID)
	if err != nil {
		return err
	}
	payload, err := json.Marshal(stats)
	if err != nil {
		return fmt.Errorf("encode run round stats: %w", err)
	}
	if _, err := s.redis.Do(ctx, "HSET", key, "roundStats", string(payload)); err != nil {
		return fmt.Errorf("save run round stats: %w", err)
	}
	return nil
}

// GetRoundStats returns the persisted summaries in execution order.
func (s *Store) GetRoundStats(ctx context.Context, runID string) ([]RoundStats, error) {
	meta, err := s.GetRunMeta(ctx, runID)
	if err != nil {
		return nil, err
	}
	sort.Slice(meta.RoundStats, func(i, j int) bool { return meta.RoundStats[i].Round < meta.RoundStats[j].Round })
	return meta.RoundStats, nil
}

func (s *Store) ActiveRuns(ctx context.Context) ([]string, error) {
	result, err := s.redis.Do(ctx, "SMEMBERS", s.ActiveRunsKey())
	if err != nil {
		return nil, fmt.Errorf("list active runs: %w", err)
	}
	return result.Strings(), nil
}

func (s *Store) RunChannels(ctx context.Context, runID string) ([]string, error) {
	key, err := s.RunChannelsKey(runID)
	if err != nil {
		return nil, err
	}
	result, err := s.redis.Do(ctx, "SMEMBERS", key)
	if err != nil {
		return nil, fmt.Errorf("list run channels: %w", err)
	}
	return result.Strings(), nil
}

// SizeKey returns the cardinality of an explicitly scoped queue key. It is
// used by the Run-aware producer for the same high-water backpressure policy
// as the legacy channel queues.
func (s *Store) SizeKey(ctx context.Context, key string) (int64, error) {
	if strings.TrimSpace(key) == "" {
		return 0, fmt.Errorf("queue key is empty")
	}
	result, err := s.redis.Do(ctx, "ZCARD", key)
	if err != nil {
		return 0, fmt.Errorf("get task count: %w", err)
	}
	return result.Int64(), nil
}

// ClaimRun claims due members from the current round for one channel.
func (s *Store) ClaimRun(ctx context.Context, runID, chanKey string, now, leaseUntil time.Time, limit int) ([]ClaimedTask, error) {
	if limit <= 0 {
		return nil, nil
	}
	if now.IsZero() {
		now = time.Now()
	}
	if leaseUntil.IsZero() || !leaseUntil.After(now) {
		return nil, fmt.Errorf("leaseUntil must be after now")
	}
	meta, err := s.GetRunMeta(ctx, runID)
	if err != nil {
		return nil, err
	}
	activeKey, _ := s.RoundActiveKey(runID, meta.CurrentRound, chanKey)
	leaseKey, _ := s.RoundLeaseKey(runID, meta.CurrentRound, chanKey)
	metaKey, _ := s.RunMetaKey(runID)
	stateKey, _ := s.TaskStateKey(runID)
	result, err := s.redis.Eval(ctx, claimRunScript, 4, []string{activeKey, leaseKey, metaKey, stateKey}, []any{now.UnixMilli(), leaseUntil.UnixMilli(), limit, s.owner, meta.CurrentRound, now.UTC().Format(time.RFC3339Nano)})
	if err != nil {
		return nil, fmt.Errorf("claim run tasks: %w", err)
	}
	values := result.Vars()
	if len(values)%2 != 0 {
		return nil, fmt.Errorf("claim returned odd number of values: %d", len(values))
	}
	tasks := make([]ClaimedTask, 0, len(values)/2)
	for i := 0; i < len(values); i += 2 {
		member := values[i].String()
		var task Task
		if err := json.Unmarshal([]byte(member), &task); err != nil {
			return nil, fmt.Errorf("decode claimed run task: %w", err)
		}
		tasks = append(tasks, ClaimedTask{Task: task, Member: member, LeaseToken: values[i+1].String(), LeaseUntil: leaseUntil.UnixMilli(), Round: meta.CurrentRound})
	}
	return tasks, nil
}

// RenewRunBatch extends leases for a Run-scoped claim. It mirrors the legacy
// RenewBatch ownership protocol while keeping the round and run hash-tagged
// keys together for Redis Cluster.
func (s *Store) RenewRunBatch(ctx context.Context, tasks []ClaimedTask, leaseUntil time.Time) (int, error) {
	if len(tasks) == 0 {
		return 0, nil
	}
	if leaseUntil.IsZero() {
		return 0, fmt.Errorf("leaseUntil is zero")
	}
	first := tasks[0]
	if first.Round <= 0 {
		return 0, fmt.Errorf("run task round is missing")
	}
	activeKey, err := s.RoundActiveKey(first.Task.RunID, first.Round, first.Task.ChanKey)
	if err != nil {
		return 0, err
	}
	leaseKey, err := s.RoundLeaseKey(first.Task.RunID, first.Round, first.Task.ChanKey)
	if err != nil {
		return 0, err
	}
	args := make([]any, 0, len(tasks)*2+1)
	args = append(args, leaseUntil.UnixMilli())
	for _, task := range tasks {
		if task.Round != first.Round || task.Task.RunID != first.Task.RunID || task.Task.ChanKey != first.Task.ChanKey {
			return 0, fmt.Errorf("renew run batch contains multiple runs, rounds, or channels")
		}
		args = append(args, task.Member, task.LeaseToken)
	}
	result, err := s.redis.Eval(ctx, renewBatchScript, 2, []string{activeKey, leaseKey}, args)
	if err != nil {
		return 0, fmt.Errorf("renew run task batch lease: %w", err)
	}
	return result.Int(), nil
}

// FinalizeRunTask atomically removes a lease and updates success/failure state.
// A non-success outcome with explicit Retry set is deferred to the next
// round; Retryable alone is reserved for lease recovery. Permanent outcomes
// are written to the final failure hash.
func (s *Store) FinalizeRunTask(ctx context.Context, claimed ClaimedTask, outcome TaskOutcome, maxRounds ...int) (bool, error) {
	count, err := s.FinalizeRunTasks(ctx, []ClaimedTask{claimed}, []TaskOutcome{outcome}, maxRounds...)
	return count == 1, err
}

// FinalizeRunTasks applies a whole claimed batch in one ownership-checked Lua
// transition. Every task must belong to the same run/round/channel.
func (s *Store) FinalizeRunTasks(ctx context.Context, claimed []ClaimedTask, outcomes []TaskOutcome, maxRounds ...int) (int, error) {
	if len(claimed) == 0 {
		return 0, nil
	}
	if len(claimed) != len(outcomes) {
		return 0, fmt.Errorf("claimed and outcome lengths differ")
	}
	first := claimed[0]
	if first.Task.RunID == "" {
		return 0, fmt.Errorf("runId is empty")
	}
	runID := first.Task.RunID
	round := first.Round
	if round <= 0 {
		round = 1
	}
	activeKey, _ := s.RoundActiveKey(runID, round, first.Task.ChanKey)
	leaseKey, _ := s.RoundLeaseKey(runID, round, first.Task.ChanKey)
	metaKey, _ := s.RunMetaKey(runID)
	stateKey, _ := s.TaskStateKey(runID)
	failureKey, _ := s.FailedKey(runID)
	waitingKey, _ := s.RoundWaitingKey(runID, round+1, first.Task.ChanKey)
	keys := []string{activeKey, leaseKey, metaKey, stateKey, waitingKey, failureKey}
	now := time.Now().UTC().Format(time.RFC3339Nano)
	args := make([]any, 0, len(claimed)*9+2)
	args = append(args, round, now)
	maxRound := 5
	if len(maxRounds) > 0 && maxRounds[0] > 0 {
		maxRound = maxRounds[0]
	}
	finalFailures := false
	for i, task := range claimed {
		if task.Round > 0 && task.Round != first.Round || task.Task.RunID != runID || task.Task.ChanKey != first.Task.ChanKey {
			return 0, fmt.Errorf("finalize run batch contains multiple runs, rounds, or channels")
		}
		outcome := outcomes[i]
		record := outcome.Failure
		if record == nil {
			record = failureFromOutcome(task.Task, outcome, round)
		}
		status := "succeeded"
		switch {
		case outcome.Covered:
			status = taskStatusCovered
		case !outcome.Success:
			status = "failed"
		}
		// Retryable describes lease recovery for infrastructure failures; only
		// the explicit Retry flag is allowed to cross the round barrier.
		deferTask := !outcome.Success && outcome.Retry && !outcome.BlockChannel && round < maxRound
		if !outcome.Success && !deferTask && outcome.ErrorClass == "mobile_busy" {
			record.FailureType = "mobile_busy_exhausted"
			record.Round = round
		}
		recordJSON, marshalErr := json.Marshal(record)
		if marshalErr != nil {
			return 0, fmt.Errorf("marshal failure record: %w", marshalErr)
		}
		if !deferTask && !outcome.Success && !outcome.Covered {
			finalFailures = true
		}
		args = append(args, task.Member, task.LeaseToken, task.Task.TaskID, status, outcome.Code, outcome.ErrCode, outcome.Message, string(recordJSON), boolString(deferTask))
	}
	result, err := s.redis.Eval(ctx, finalizeRunBatchScript, int64(len(keys)), keys, args)
	if err != nil {
		return 0, fmt.Errorf("finalize run task batch: %w", err)
	}
	count := result.Int()
	if count == len(claimed) && finalFailures {
		if _, err = s.redis.Do(ctx, "ZADD", s.FailedRunsKey(), time.Now().UnixMilli(), runID); err != nil {
			return 0, fmt.Errorf("index failed run: %w", err)
		}
	}
	return count, nil
}

func (s *Store) finalizeRunTaskLegacy(ctx context.Context, claimed ClaimedTask, outcome TaskOutcome) (bool, error) {
	if claimed.Task.RunID == "" {
		return false, fmt.Errorf("runId is empty")
	}
	runID := claimed.Task.RunID
	round := claimed.Round
	if round <= 0 {
		round = 1
	}
	activeKey, _ := s.RoundActiveKey(runID, round, claimed.Task.ChanKey)
	leaseKey, _ := s.RoundLeaseKey(runID, round, claimed.Task.ChanKey)
	metaKey, _ := s.RunMetaKey(runID)
	stateKey, _ := s.TaskStateKey(runID)
	failureKey, _ := s.FailedKey(runID)
	waitingKey, _ := s.RoundWaitingKey(runID, round+1, claimed.Task.ChanKey)
	record := outcome.Failure
	if record == nil {
		record = failureFromOutcome(claimed.Task, outcome, round)
	}
	status := "succeeded"
	switch {
	case outcome.Covered:
		status = taskStatusCovered
	case !outcome.Success:
		status = "failed"
	}
	deferTask := !outcome.Success && outcome.Retry && !outcome.BlockChannel && round < 5
	if !outcome.Success && !deferTask && outcome.ErrorClass == "mobile_busy" {
		record.FailureType = "mobile_busy_exhausted"
		record.Round = round
	}
	recordJSON, _ := json.Marshal(record)
	args := []any{claimed.Member, claimed.LeaseToken, claimed.Task.TaskID, status, round, outcome.Code, outcome.ErrCode, outcome.Message, string(recordJSON), boolString(deferTask), time.Now().UTC().Format(time.RFC3339Nano)}
	keys := []string{activeKey, leaseKey, metaKey, stateKey, waitingKey, failureKey}
	result, err := s.redis.Eval(ctx, finalizeRunScript, int64(len(keys)), keys, args)
	if err != nil {
		return false, fmt.Errorf("finalize run task: %w", err)
	}
	if result.Int() != 1 {
		return false, nil
	}
	if !deferTask && !outcome.Success && !outcome.Covered {
		// The failed-run index is global and cannot be part of the run-key Lua
		// transaction on Redis Cluster. The per-task failed hash and counters
		// above are the atomic state; this index is only a discovery aid.
		if _, err = s.redis.Do(ctx, "ZADD", s.FailedRunsKey(), time.Now().UnixMilli(), runID); err != nil {
			return false, fmt.Errorf("index failed run: %w", err)
		}
	}
	return true, nil
}

func (s *Store) DeferRunTask(ctx context.Context, claimed ClaimedTask, failure FailureRecord) (bool, error) {
	return s.FinalizeRunTask(ctx, claimed, TaskOutcome{Retry: true, Retryable: true, Failure: &failure, Code: failure.Code, ErrCode: failure.ErrCode, Message: failure.Message, FailureType: failure.FailureType})
}

func (s *Store) FailRunTask(ctx context.Context, claimed ClaimedTask, failure FailureRecord) (bool, error) {
	return s.FinalizeRunTask(ctx, claimed, TaskOutcome{Failure: &failure, Code: failure.Code, ErrCode: failure.ErrCode, Message: failure.Message, FailureType: failure.FailureType})
}

// AdvanceRunRound atomically promotes waiting members only when the producer
// has completed and both active and inflight counts are zero.
func (s *Store) AdvanceRunRound(ctx context.Context, runID string, maxRounds ...int) (bool, error) {
	meta, err := s.GetRunMeta(ctx, runID)
	if err != nil {
		return false, err
	}
	if meta.CurrentRound <= 0 {
		meta.CurrentRound = 1
	}
	channels, err := s.RunChannels(ctx, runID)
	if err != nil {
		return false, err
	}
	metaKey, _ := s.RunMetaKey(runID)
	stateKey, _ := s.TaskStateKey(runID)
	keys := []string{metaKey, stateKey}
	for _, ch := range channels {
		w, _ := s.RoundWaitingKey(runID, meta.CurrentRound+1, ch)
		a, _ := s.RoundActiveKey(runID, meta.CurrentRound+1, ch)
		keys = append(keys, w, a)
	}
	maxRound := 5
	if len(maxRounds) > 0 && maxRounds[0] > 0 {
		maxRound = maxRounds[0]
	}
	args := []any{meta.CurrentRound, maxRound, time.Now().UTC().Format(time.RFC3339Nano)}
	result, err := s.redis.Eval(ctx, advanceRoundScript, int64(len(keys)), keys, args)
	if err != nil {
		return false, fmt.Errorf("advance run round: %w", err)
	}
	return result.Int() == 1, nil
}

// CompleteRunIfIdle closes a run when no active/inflight/waiting work remains.
func (s *Store) CompleteRunIfIdle(ctx context.Context, runID string, retention ...time.Duration) (bool, error) {
	meta, err := s.GetRunMeta(ctx, runID)
	if err != nil {
		return false, err
	}
	channels, err := s.RunChannels(ctx, runID)
	if err != nil {
		return false, err
	}
	metaKey, _ := s.RunMetaKey(runID)
	failedKey, _ := s.FailedKey(runID)
	stateKey, _ := s.TaskStateKey(runID)
	channelsKey, _ := s.RunChannelsKey(runID)
	keys := []string{metaKey, failedKey, stateKey, channelsKey}
	for _, ch := range channels {
		a, _ := s.RoundActiveKey(runID, meta.CurrentRound, ch)
		l, _ := s.RoundLeaseKey(runID, meta.CurrentRound, ch)
		w, _ := s.RoundWaitingKey(runID, meta.CurrentRound+1, ch)
		keys = append(keys, a, l, w)
	}
	keep := 30 * 24 * time.Hour
	if len(retention) > 0 && retention[0] > 0 {
		keep = retention[0]
	}
	result, err := s.redis.Eval(ctx, completeRunScript, int64(len(keys)), keys, []any{time.Now().UTC().Format(time.RFC3339Nano), int64(keep.Milliseconds())})
	if err != nil {
		return false, fmt.Errorf("complete run: %w", err)
	}
	if result.Int() != 1 {
		return false, nil
	}
	// Keep global indexes out of the run-scoped Lua transaction (see
	// CreateRun). The metadata transition above is the authoritative CAS.
	if _, err = s.redis.Do(ctx, "SREM", s.ActiveRunsKey(), runID); err != nil {
		return false, fmt.Errorf("remove completed run index: %w", err)
	}
	if meta.FailedCount > 0 {
		if _, err = s.redis.Do(ctx, "ZADD", s.FailedRunsKey(), time.Now().UnixMilli(), runID); err != nil {
			return false, fmt.Errorf("index failed run: %w", err)
		}
	}
	if s.archiveEnabled {
		if _, err = s.redis.Do(ctx, "ZADD", s.prefix+"archive:pending", time.Now().UnixMilli(), runID); err != nil {
			return false, fmt.Errorf("index archive pending run: %w", err)
		}
	}
	return true, nil
}

// BlockRunScoped is the run-queue counterpart of the legacy BlockRun API. It
// removes only this run's active, leased and waiting members on a channel and
// marks the channel blocked in run metadata. BlockRunRun is kept as a readable
// alias for callers that prefer runID-first argument ordering.
func (s *Store) BlockRunScoped(ctx context.Context, runID, chanKey string) (int64, error) {
	meta, err := s.GetRunMeta(ctx, runID)
	if err != nil {
		return 0, err
	}
	active, _ := s.RoundActiveKey(runID, meta.CurrentRound, chanKey)
	lease, _ := s.RoundLeaseKey(runID, meta.CurrentRound, chanKey)
	waiting, _ := s.RoundWaitingKey(runID, meta.CurrentRound+1, chanKey)
	metaKey, _ := s.RunMetaKey(runID)
	stateKey, _ := s.TaskStateKey(runID)
	result, err := s.redis.Eval(ctx, blockRunScopedScript, 5, []string{active, lease, waiting, metaKey, stateKey}, []any{runID, chanKey, time.Now().UTC().Format(time.RFC3339Nano)})
	if err != nil {
		return 0, fmt.Errorf("block run channel: %w", err)
	}
	return result.Int64(), nil
}

func (s *Store) BlockRunRun(ctx context.Context, runID, chanKey string) (int64, error) {
	return s.BlockRunScoped(ctx, runID, chanKey)
}

// BlockRunScopedAsFailed preserves a terminal failure for every unfinished
// status task on a blocked channel. A negative count means the triggering
// lease was lost; an obsolete consumer must not block another owner's work.
func (s *Store) BlockRunScopedAsFailed(ctx context.Context, trigger ClaimedTask, outcome TaskOutcome) (int64, error) {
	if trigger.Round <= 0 || trigger.Task.RunID == "" || trigger.Task.ChanKey == "" || trigger.LeaseToken == "" {
		return 0, gerror.New("channel block requires a run-scoped triggering lease")
	}
	var (
		runID   = trigger.Task.RunID
		chanKey = trigger.Task.ChanKey
		round   = trigger.Round
	)
	active, _ := s.RoundActiveKey(runID, round, chanKey)
	lease, _ := s.RoundLeaseKey(runID, round, chanKey)
	waiting, _ := s.RoundWaitingKey(runID, round+1, chanKey)
	metaKey, _ := s.RunMetaKey(runID)
	stateKey, _ := s.TaskStateKey(runID)
	failureKey, _ := s.FailedKey(runID)
	outcome.FailureType = "channel_blocked"
	record, err := json.Marshal(failureFromOutcome(trigger.Task, outcome, round))
	if err != nil {
		return 0, gerror.Wrap(err, "marshal channel block failure")
	}
	result, err := s.redis.Eval(ctx, blockRunScopedScript, 6,
		[]string{active, lease, waiting, metaKey, stateKey, failureKey},
		[]any{runID, chanKey, time.Now().UTC().Format(time.RFC3339Nano), "1", trigger.Member, trigger.LeaseToken, string(record)})
	if err != nil {
		return 0, gerror.Wrap(err, "fail blocked run channel")
	}
	if result.Int64() > 0 {
		if _, err = s.redis.Do(ctx, "ZADD", s.FailedRunsKey(), time.Now().UnixMilli(), runID); err != nil {
			return 0, gerror.Wrap(err, "index blocked run failures")
		}
	}
	return result.Int64(), nil
}

func boolString(v bool) string {
	if v {
		return "1"
	}
	return "0"
}
func parseBool(v string) bool      { return v == "1" || strings.EqualFold(v, "true") }
func parseInt(v string) int        { n, _ := strconv.Atoi(v); return n }
func parseInt64(v string) int64    { n, _ := strconv.ParseInt(v, 10, 64); return n }
func parseTime(v string) time.Time { t, _ := time.Parse(time.RFC3339Nano, v); return t }
func parseOptionalTime(v string) *time.Time {
	if v == "" {
		return nil
	}
	t, err := time.Parse(time.RFC3339Nano, v)
	if err != nil {
		return nil
	}
	return &t
}

func failureFromOutcome(task Task, o TaskOutcome, round int) *FailureRecord {
	now := time.Now()
	typ := o.FailureType
	if typ == "" {
		typ = o.ErrorClass
	}
	carrier := "generic"
	if model.IsTelecomTemplate(task.TempKey) {
		carrier = "telecom"
	} else {
		switch strings.ToLower(strings.TrimSpace(task.TempKey)) {
		// The list must stay in sync with carrierMobile's template set in
		// internal/consumer/zset/consumer.go; a missing entry silently labels the
		// failure generic, which breaks carrier-based failure review and replay.
		case "sim_yd_v5", "sim_yd_mock_v5", "sim_onelink_tengyu_v1", "sim_onelink_xuzhou_mobile_v1", "sim_onelink_lelian_v1", "sim_onelink_beixin_v1", "sim_onelink_qianxun_v1", "sim_onelink_henan_xinjing_v1", "sim_onelink_daihe_v1", "lanshang_8000", "lanshang_8003":
			carrier = "mobile"
		}
	}
	return &FailureRecord{Version: 1, TaskID: task.TaskID, OriginalRunID: task.RunID, LatestRunID: task.RunID, QueryMonth: task.QueryMonth, Operation: task.Operation, CardID: task.CardID, ICCID: task.ICCID, MSISDN: task.MSISDN, IMSI: task.IMSI, ChanKey: task.ChanKey, TempKey: task.TempKey, PackageCode: task.PackageCode, Carrier: carrier, FailureType: typ, Round: round, Attempts: round, Code: o.Code, ErrCode: o.ErrCode, Message: o.Message, ReviewStatus: "pending", FirstFailedAt: now, LastFailedAt: now}
}

const createRunScript = `
if redis.call('EXISTS', KEYS[1]) == 0 then
 redis.call('HSET', KEYS[1], 'runId', ARGV[1], 'operation', ARGV[2], 'queryMonth', ARGV[3], 'status', ARGV[4], 'producerDone', ARGV[5], 'currentRound', ARGV[6], 'activeCount', ARGV[7], 'inflightCount', ARGV[8], 'nextRoundCount', ARGV[9], 'successCount', ARGV[10], 'failedCount', ARGV[11], 'createdAt', ARGV[12])
 return 1
end
return 2
`
const enqueueRunScript = `
if redis.call('EXISTS', KEYS[1]) == 0 then return 0 end
if redis.call('HGET', KEYS[1], 'status') == 'completed' or redis.call('HGET', KEYS[1], 'status') == 'completed_with_failures' then return 0 end
if redis.call('HGET', KEYS[1], 'producerDone') == '1' then return 0 end
if redis.call('HGET', KEYS[1], 'blocked:' .. ARGV[1]) == '1' then return 0 end
redis.call('SADD', KEYS[2], ARGV[1])
local added = 0
for i=5,#ARGV,2 do
 local taskID=ARGV[i]; local member=ARGV[i+1]
 if redis.call('HEXISTS', KEYS[4], taskID) == 0 then
  redis.call('ZADD', KEYS[3], ARGV[2], member)
  redis.call('HSET', KEYS[4], taskID, cjson.encode({round=tonumber(ARGV[3]),status='queued',attempts=0,updatedAt=ARGV[4]}))
  redis.call('HINCRBY', KEYS[1], 'activeCount', 1)
  added=added+1
 end
end
return added
`
const markProducerDoneScript = `
if redis.call('EXISTS', KEYS[1]) == 0 then return 0 end
local status=redis.call('HGET',KEYS[1],'status'); if status=='completed' or status=='completed_with_failures' then return 2 end
redis.call('HSET', KEYS[1], 'producerDone', '1', 'producerCompletedAt', ARGV[1])
if redis.call('HGET', KEYS[1], 'status') == 'producing' then redis.call('HSET', KEYS[1], 'status', 'running') end
return 1
`

// markProducerAbandonedScript promotes only a run whose producer provably never
// finished: "producing" or channel-blocked "running", producerDone not set,
// and not a replay target. Callers establish ownership of the producer lock.
const markProducerAbandonedScript = `
if redis.call('EXISTS', KEYS[1]) == 0 then return 0 end
local status = redis.call('HGET', KEYS[1], 'status')
if status ~= 'producing' and status ~= 'running' then return 0 end
if redis.call('HGET', KEYS[1], 'producerDone') == '1' then return 0 end
local syncType = redis.call('HGET', KEYS[1], 'syncType')
if syncType == 'replay' then return 0 end
redis.call('HSET', KEYS[1], 'producerDone', '1', 'producerCompletedAt', ARGV[1], 'abandonedAt', ARGV[1], 'status', 'running')
return 1
`
const claimRunScript = `
local status=redis.call('HGET',KEYS[3],'status')
if status=='completed' or status=='completed_with_failures' then return {} end
local members=redis.call('ZRANGEBYSCORE',KEYS[1],'-inf',ARGV[1],'LIMIT',0,ARGV[3]); local result={}; local serverTime=redis.call('TIME')
for i,member in ipairs(members) do
 local previous=redis.call('HGET',KEYS[2],member)
 if previous then redis.call('HINCRBY',KEYS[3],'inflightCount',-1) end
 local token=ARGV[4]..':'..serverTime[1]..':'..serverTime[2]..':'..i
 redis.call('ZADD',KEYS[1],ARGV[2],member); redis.call('HSET',KEYS[2],member,token)
 local decoded=cjson.decode(member); local taskID=decoded.taskId
 local old=redis.call('HGET',KEYS[4],taskID); local attempts=0
 if old then local ok,state=pcall(cjson.decode,old); if ok and state.attempts then attempts=state.attempts end end
 redis.call('HSET',KEYS[4],taskID,cjson.encode({round=tonumber(ARGV[5]),status='claimed',attempts=attempts+1,updatedAt=ARGV[6]}))
 table.insert(result,member); table.insert(result,token)
end
if #members>0 then redis.call('HINCRBY',KEYS[3],'inflightCount',#members) end
return result
`
const finalizeRunScript = `
if redis.call('HGET',KEYS[2],ARGV[1]) ~= ARGV[2] then return 0 end
redis.call('ZREM',KEYS[1],ARGV[1]); redis.call('HDEL',KEYS[2],ARGV[1]); redis.call('HINCRBY',KEYS[3],'activeCount',-1); redis.call('HINCRBY',KEYS[3],'inflightCount',-1)
local attempts=tonumber(ARGV[5])
local previous=redis.call('HGET',KEYS[4],ARGV[3])
if previous then local ok,state=pcall(cjson.decode,previous); if ok and state.attempts then attempts=tonumber(state.attempts) or attempts end end
local taskStatus=ARGV[4]
if ARGV[10]=='1' then taskStatus='waiting_next_round' end
if taskStatus=='failed' then taskStatus='failed_pending_review' end
redis.call('HSET',KEYS[4],ARGV[3], cjson.encode({round=tonumber(ARGV[5]),status=taskStatus,attempts=attempts,lastCode=tonumber(ARGV[6]),lastErrCode=ARGV[7],lastMessage=ARGV[8],updatedAt=ARGV[11]}))
if ARGV[10]=='1' then redis.call('ZADD',KEYS[5],0,ARGV[1]); redis.call('HINCRBY',KEYS[3],'nextRoundCount',1) else if ARGV[4]=='succeeded' or ARGV[4]=='covered' then redis.call('HINCRBY',KEYS[3],'successCount',1) else local failureRecord=ARGV[9]; local ok,failure=pcall(cjson.decode,failureRecord); if ok and failure then failure.round=tonumber(ARGV[5]); failure.attempts=attempts; failureRecord=cjson.encode(failure) end; redis.call('HSET',KEYS[6],ARGV[3],failureRecord); redis.call('HINCRBY',KEYS[3],'failedCount',1) end end
return 1
`
const finalizeRunBatchScript = `
local count=(#ARGV-2)/9
for i=3,#ARGV,9 do
 if redis.call('HGET',KEYS[2],ARGV[i]) ~= ARGV[i+1] then return 0 end
end
for i=3,#ARGV,9 do
 local member=ARGV[i]; local taskID=ARGV[i+2]; local status=ARGV[i+3]; local code=ARGV[i+4]; local errCode=ARGV[i+5]; local message=ARGV[i+6]; local record=ARGV[i+7]; local deferTask=ARGV[i+8]
 redis.call('ZREM',KEYS[1],member); redis.call('HDEL',KEYS[2],member)
 redis.call('HINCRBY',KEYS[3],'activeCount',-1); redis.call('HINCRBY',KEYS[3],'inflightCount',-1)
 local attempts=tonumber(ARGV[1]); local previous=redis.call('HGET',KEYS[4],taskID)
 if previous then local ok,state=pcall(cjson.decode,previous); if ok and state.attempts then attempts=tonumber(state.attempts) or attempts end end
 local taskStatus=status; if deferTask=='1' then taskStatus='waiting_next_round' elseif taskStatus=='failed' then taskStatus='failed_pending_review' end
 redis.call('HSET',KEYS[4],taskID,cjson.encode({round=tonumber(ARGV[1]),status=taskStatus,attempts=attempts,lastCode=tonumber(code),lastErrCode=errCode,lastMessage=message,updatedAt=ARGV[2]}))
 if deferTask=='1' then redis.call('ZADD',KEYS[5],0,member); redis.call('HINCRBY',KEYS[3],'nextRoundCount',1)
 elseif status=='succeeded' or status=='covered' then redis.call('HINCRBY',KEYS[3],'successCount',1)
 else local ok,failure=pcall(cjson.decode,record); if ok and failure then failure.round=tonumber(ARGV[1]); failure.attempts=attempts; record=cjson.encode(failure) end; redis.call('HSET',KEYS[6],taskID,record); redis.call('HINCRBY',KEYS[3],'failedCount',1) end
end
return count
`
const advanceRoundScript = `
if redis.call('HGET',KEYS[1],'producerDone') ~= '1' then return 0 end
local status=redis.call('HGET',KEYS[1],'status'); if status=='completed' or status=='completed_with_failures' then return 0 end
if tonumber(redis.call('HGET',KEYS[1],'currentRound') or '0') ~= tonumber(ARGV[1]) then return 0 end
if tonumber(ARGV[1]) >= tonumber(ARGV[2]) then return 0 end
if tonumber(redis.call('HGET',KEYS[1],'activeCount') or '0') ~= 0 or tonumber(redis.call('HGET',KEYS[1],'inflightCount') or '0') ~= 0 then return 0 end
local moved=0; local nextRound=tonumber(ARGV[1])+1
for i=3,#KEYS,2 do local waiting=KEYS[i]; local active=KEYS[i+1]; local members=redis.call('ZRANGE',waiting,0,-1,'WITHSCORES'); for j=1,#members,2 do redis.call('ZADD',active,members[j+1],members[j]); local payload=cjson.decode(members[j]); local attempts=nextRound-1; local previous=redis.call('HGET',KEYS[2],payload.taskId); if previous then local ok,state=pcall(cjson.decode,previous); if ok and state.attempts then attempts=tonumber(state.attempts) or attempts end end; redis.call('HSET',KEYS[2],payload.taskId,cjson.encode({round=nextRound,status='queued',attempts=attempts,updatedAt=ARGV[3]})); moved=moved+1 end; redis.call('DEL',waiting); end
if moved==0 then return 0 end
redis.call('HSET',KEYS[1],'currentRound',nextRound,'activeCount',moved,'inflightCount',0,'nextRoundCount',0)
return 1
`
const completeRunScript = `
if redis.call('HGET',KEYS[1],'producerDone') ~= '1' then return 0 end
local existingStatus=redis.call('HGET',KEYS[1],'status'); if existingStatus=='completed' or existingStatus=='completed_with_failures' then return 0 end
if tonumber(redis.call('HGET',KEYS[1],'activeCount') or '0') ~= 0 or tonumber(redis.call('HGET',KEYS[1],'inflightCount') or '0') ~= 0 or tonumber(redis.call('HGET',KEYS[1],'nextRoundCount') or '0') ~= 0 then return 0 end
local failed=tonumber(redis.call('HGET',KEYS[1],'failedCount') or '0'); local status='completed'; if failed>0 then status='completed_with_failures'; end
redis.call('HSET',KEYS[1],'status',status,'completedAt',ARGV[1]); redis.call('PEXPIRE',KEYS[1],ARGV[2]); redis.call('PEXPIRE',KEYS[2],ARGV[2]); for i=3,#KEYS do redis.call('PEXPIRE',KEYS[i],ARGV[2]); end; return 1
`

const blockRunScopedScript = `
local status=redis.call('HGET',KEYS[4],'status')
if status=='completed' or status=='completed_with_failures' then return 0 end
local countFailures=ARGV[4]=='1'
if countFailures and redis.call('HGET',KEYS[2],ARGV[5]) ~= ARGV[6] then return -1 end
local removed = 0
local leased = 0
local waiting = 0
local function markBlocked(member)
  local ok,payload=pcall(cjson.decode,member)
  if not ok or not payload or not payload.taskId then return end
  local round=0; local attempts=0; local previous=redis.call('HGET',KEYS[5],payload.taskId)
  if previous then local stateOK,state=pcall(cjson.decode,previous); if stateOK and state then round=tonumber(state.round) or 0; attempts=tonumber(state.attempts) or 0 end end
  redis.call('HSET',KEYS[5],payload.taskId,cjson.encode({round=round,status='channel_blocked',attempts=attempts,updatedAt=ARGV[3]}))
  if countFailures then
    local record=cjson.decode(ARGV[7])
    record.taskId=payload.taskId; record.cardId=payload.cardId; record.iccid=payload.iccid
    record.msisdn=payload.msisdn; record.imsi=payload.imsi; record.operation=payload.operation
    record.round=round; record.attempts=attempts
    redis.call('HSET',KEYS[6],payload.taskId,cjson.encode(record))
    redis.call('HINCRBY',KEYS[4],'failedCount',1)
  end
end
local activeMembers = redis.call('ZRANGE', KEYS[1], 0, -1)
for _, member in ipairs(activeMembers) do
  if string.find(member, '"runId":"' .. ARGV[1] .. '"', 1, true) then
    redis.call('ZREM', KEYS[1], member)
    if redis.call('HDEL', KEYS[2], member) == 1 then leased = leased + 1 end
    markBlocked(member)
    removed = removed + 1
  end
end
local waitingMembers = redis.call('ZRANGE', KEYS[3], 0, -1)
for _, member in ipairs(waitingMembers) do
  redis.call('ZREM', KEYS[3], member)
  markBlocked(member)
  waiting = waiting + 1
end
redis.call('HINCRBY', KEYS[4], 'activeCount', -removed)
if leased > 0 then redis.call('HINCRBY', KEYS[4], 'inflightCount', -leased) end
if waiting > 0 then redis.call('HINCRBY', KEYS[4], 'nextRoundCount', -waiting) end
redis.call('HSET', KEYS[4], 'blocked:' .. ARGV[2], '1', 'status', 'running')
return removed + waiting
`