iotlcm/internal/producer/imei/producer.go

package imei

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

	"github.com/gogf/gf/v2/database/gredis"
	"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"
	queue "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
)

// FullSyncOperation is recorded on the run and used by SyncKindForOperation to
// classify the run for the admin monitor.
const FullSyncOperation = "imei_full_sync"

// scopeSampleLimit bounds how many channel names are copied into run metadata.
const scopeSampleLimit = 10

// productionStats mirrors the status producer so every full-sync path reports a
// single comparable summary line.
type productionStats struct {
	pages               int64
	scanned             int64
	enqueued            int64
	skippedEmptyChan    int64
	skippedNoTemplate   int64
	skippedSyncDisabled int64
	skippedOutOfScope   int64
	skippedNoImei       int64
	scopeChannels       int
	channels            map[string]struct{}
}

func newProductionStats(scopeChannels int) productionStats {
	return productionStats{channels: make(map[string]struct{}), scopeChannels: scopeChannels}
}

func (s *productionStats) log(ctx context.Context, status, runID, stage, reason string, startedAt time.Time, maxID int64, err error) {
	message := fmt.Sprintf("event=imei_queue_producer status=%s runId=%s snapshotMaxId=%d pages=%d scanned=%d enqueued=%d skippedEmptyChan=%d skippedNoTemplate=%d skippedSyncDisabled=%d skippedOutOfScope=%d skippedNoImei=%d channelCount=%d scopeChannels=%d durationMs=%d", status, runID, maxID, s.pages, s.scanned, s.enqueued, s.skippedEmptyChan, s.skippedNoTemplate, s.skippedSyncDisabled, s.skippedOutOfScope, s.skippedNoImei, len(s.channels), s.scopeChannels, 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())
	}
	switch status {
	case "failed":
		g.Log().Error(ctx, message)
	case "skipped":
		g.Log().Warning(ctx, message)
	default:
		g.Log().Info(ctx, message)
	}
}

type Config struct {
	PageSize            int
	RunLockKey          string
	RunLockTTL          time.Duration
	PollInterval        time.Duration
	GlobalHighWaterMark int64
	// LeaseRedisConfig is required only for Sentinel discovery. Standalone
	// settings come from the actual adapter; Cluster is rejected in New.
	LeaseRedisConfig *gredis.Config
}

func (c Config) withDefaults() Config {
	if c.PageSize <= 0 {
		c.PageSize = 1000
	}
	if c.RunLockKey == "" {
		c.RunLockKey = "imei:producer:active"
	}
	if c.RunLockTTL <= 0 {
		c.RunLockTTL = 5 * time.Minute
	}
	if c.PollInterval <= 0 {
		c.PollInterval = time.Second
	}
	return c
}

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

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

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

// DatabaseScanner uses keyset pagination so a large IMEI run does not hold the
// entire card table in memory and does not degrade through OFFSET scans.
type DatabaseScanner struct {
	metadata map[string]channelMetadata
}

func (s *DatabaseScanner) SnapshotMaxID(ctx context.Context) (int64, error) {
	var card entity.TSimCardInfo
	columns := dao.TSimCardInfo.Columns()
	if err := dao.TSimCardInfo.Ctx(ctx).
		Fields(columns.Id).
		OrderDesc(columns.Id).
		Limit(1).
		Scan(&card); err != nil {
		return 0, fmt.Errorf("read imei 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 s.metadata == nil {
		var rows []channelMetadata
		if err := g.DB().Model("t_sim_channels").Ctx(ctx).Fields("chan_key,temp_key,sync_enabled,operator").Scan(&rows); err != nil {
			return nil, fmt.Errorf("scan imei channel metadata: %w", err)
		}
		s.metadata = make(map[string]channelMetadata, len(rows))
		for _, row := range rows {
			s.metadata[row.ChanKey] = row
		}
	}
	columns := dao.TSimCardInfo.Columns()
	var rows []entity.TSimCardInfo
	if err := dao.TSimCardInfo.Ctx(ctx).
		Fields(columns.Id, columns.Iccid, columns.Msisdn, columns.Imsi, columns.ChanKey, columns.Imei).
		WhereGT(columns.Id, lastID).
		WhereLTE(columns.Id, maxID).
		OrderAsc(columns.Id).
		Limit(pageSize).
		Scan(&rows); err != nil {
		return nil, fmt.Errorf("scan imei cards: %w", err)
	}
	cards := make([]Card, 0, len(rows))
	for _, row := range rows {
		if row.Id <= 0 || row.Iccid == "" || row.ChanKey == "" {
			// Keep the page cursor moving even when a row cannot become a
			// runnable task. The producer advances by the highest returned ID.
			if row.Id > 0 {
				cards = append(cards, Card{ID: row.Id})
			}
			continue
		}
		metadata := s.metadata[row.ChanKey]
		cards = append(cards, Card{ID: row.Id, ICCID: row.Iccid, MSISDN: row.Msisdn, IMSI: row.Imsi, ChanKey: row.ChanKey, TempKey: metadata.TempKey, Operator: metadata.Operator, SyncEnabled: metadata.SyncEnabled, Imei: row.Imei})
	}
	return cards, nil
}

type Producer struct {
	scanner      Scanner
	store        *queue.Store
	config       Config
	running      atomic.Int32
	owner        string
	leaseFactory leaseClientFactory
}

var ErrAlreadyRunning = fmt.Errorf("imei 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.
var ErrRunFinished = fmt.Errorf("imei 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 strings.Contains(err.Error(), "not found") {
			return false, nil
		}
		return false, err
	}
	return meta.Status == "completed" || meta.Status == "completed_with_failures", nil
}

func New(scanner Scanner, store *queue.Store, config Config) (*Producer, error) {
	if scanner == nil || store == nil {
		return nil, fmt.Errorf("imei producer dependencies are incomplete")
	}
	config = config.withDefaults()
	factory, err := newLeaseClientFactory(store.RedisClient(), config.LeaseRedisConfig, leaseIOBudget(config.RunLockTTL))
	if err != nil {
		return nil, err
	}
	return &Producer{scanner: scanner, store: store, config: config, owner: newOwner(), leaseFactory: factory}, nil
}

func leaseIOBudget(ttl time.Duration) time.Duration {
	return min(time.Second, ttl/6)
}

func (p *Producer) Produce(ctx context.Context, message model.ImeiQueueStartMessage) (err error) {
	startedAt := time.Now()
	if message.RunID == "" {
		return fmt.Errorf("imei sync runId is empty")
	}
	operation := message.Operation
	if operation == "" {
		operation = FullSyncOperation
	}
	scope, err := model.NormalizeScope(message.Scope)
	if err != nil {
		failedStats := newProductionStats(0)
		failedStats.log(ctx, "failed", message.RunID, "validate", "invalid_scope", startedAt, 0, err)
		return err
	}
	stats := newProductionStats(scope.ChannelCount())
	if !p.running.CompareAndSwap(0, 1) {
		stats.log(ctx, "skipped", message.RunID, "precheck", "already_running", startedAt, 0, nil)
		return ErrAlreadyRunning
	}
	defer p.running.Store(0)
	leaseClient, closeLeaseClient, err := p.leaseFactory()
	if err != nil {
		return err
	}
	defer closeLeaseClient()
	leaseStore, err := queue.New(leaseClient, p.store.Prefix())
	if err != nil {
		return err
	}
	// A redelivered start message must not resurrect a run that already
	// finished; that would create a second producer pass over the same RunID.
	if finished, checkErr := p.runAlreadyFinished(ctx, message.RunID); checkErr != nil {
		stats.log(ctx, "failed", message.RunID, "run_state", "", startedAt, 0, checkErr)
		return checkErr
	} else if finished {
		stats.log(ctx, "skipped", message.RunID, "run_state", "run_finished", startedAt, 0, nil)
		return ErrRunFinished
	}
	// Tokens are per lease, not per Producer: a delayed command from an earlier
	// attempt must never match a later lease on this same Producer instance.
	p.owner = newOwner()
	acquireCtx, cancelAcquire := context.WithTimeout(ctx, leaseIOBudget(p.config.RunLockTTL))
	acquired, err := leaseStore.AcquireLockOrTakeover(acquireCtx, p.config.RunLockKey, p.owner, message.RunID, p.config.RunLockTTL)
	cancelAcquire()
	if err != nil {
		stats.log(ctx, "failed", message.RunID, "lock", "", startedAt, 0, err)
		return err
	}
	if !acquired {
		stats.log(ctx, "skipped", message.RunID, "lock", "already_running", startedAt, 0, nil)
		return ErrAlreadyRunning
	}
	defer func() {
		releaseCtx, cancel := context.WithTimeout(context.Background(), leaseIOBudget(p.config.RunLockTTL))
		defer cancel()
		_, _ = leaseStore.ReleaseLock(releaseCtx, p.config.RunLockKey, p.owner)
	}()
	workStore, err := p.store.WithIMEIProducerLease(p.config.RunLockKey, p.owner, message.RunID)
	if err != nil {
		return err
	}
	lockCtx, cancelLock := context.WithCancelCause(ctx)
	// Stop renewal separately so normal cleanup does not become a lock-loss cause.
	renewCtx, stopRenew := context.WithCancel(lockCtx)
	renewDone := make(chan struct{})
	var (
		maxID int64
		stage = "run_init"
	)
	go func() {
		defer close(renewDone)
		p.renewLock(renewCtx, cancelLock, leaseStore)
	}()
	defer func() {
		stopRenew()
		<-renewDone
		// Join before reading the cause and releasing the lease: a failed renewal
		// must not be masked by a scanner/Redis context-cancellation error.
		if cause := context.Cause(lockCtx); cause != nil {
			err = cause
		}
		if errors.Is(err, queue.ErrProducerLeaseLost) {
			err = ErrAlreadyRunning
		}
		cancelLock(nil)
		if err != nil {
			stats.log(ctx, "failed", message.RunID, stage, "", startedAt, maxID, err)
			return
		}
		stats.log(ctx, "completed", message.RunID, "", "", startedAt, maxID, nil)
	}()
	if err = workStore.CreateRun(lockCtx, queue.RunMeta{
		RunID:               message.RunID,
		Operation:           operation,
		Status:              "producing",
		CurrentRound:        1,
		ScopeChannels:       scope.ChannelCount(),
		ScopeDigest:         scopeDigest(scope),
		ScopeChannelsSample: strings.Join(scope.ChannelSample(scopeSampleLimit), ","),
	}); err != nil {
		return err
	}
	stage = "snapshot"
	if err = lockCtx.Err(); err != nil {
		return err
	}
	maxID, err = p.scanner.SnapshotMaxID(lockCtx)
	if err != nil {
		return err
	}
	g.Log().Infof(ctx, "event=imei_queue_producer status=started runId=%s snapshotMaxId=%d scopeChannels=%d", message.RunID, maxID, scope.ChannelCount())
	lastID := int64(0)
	for lastID < maxID {
		if err = lockCtx.Err(); err != nil {
			return err
		}
		stage = "scan"
		cards, scanErr := p.scanner.Scan(lockCtx, lastID, maxID, p.config.PageSize)
		if scanErr != nil {
			return scanErr
		}
		if len(cards) == 0 {
			break
		}
		stats.pages++
		stats.scanned += int64(len(cards))
		groups := make(map[string][]queue.Task)
		for _, card := range cards {
			if err = lockCtx.Err(); err != nil {
				return err
			}
			if card.ID > lastID {
				lastID = card.ID
			}
			if card.ChanKey == "" {
				stats.skippedEmptyChan++
				continue
			}
			// Scope is applied before the template and sync_enabled checks so a
			// partial run only reports on the channels it was asked for.
			if !scope.Allows(card.ChanKey) {
				stats.skippedOutOfScope++
				continue
			}
			if card.TempKey == "" {
				stats.skippedNoTemplate++
				continue
			}
			if card.SyncEnabled {
				// sync_enabled=1(否/不参与) → skip
				stats.skippedSyncDisabled++
				continue
			}
			// skippedNoImei counts cards that already have a value. They are
			// still enqueued (this is a full re-poll, so a device swap is
			// detected), but the count makes the "how much of this pass is
			// actually filling gaps" question answerable from one log line.
			if strings.TrimSpace(card.Imei) != "" {
				stats.skippedNoImei++
			}
			stats.channels[card.ChanKey] = struct{}{}
			task := queue.Task{TaskID: queue.StableTaskID(message.RunID, operation, card.ID), RunID: message.RunID, Operation: operation, ChanKey: card.ChanKey, TempKey: card.TempKey, Operator: card.Operator, CardID: card.ID, ICCID: card.ICCID, MSISDN: card.MSISDN, IMSI: card.IMSI}
			groups[card.ChanKey] = append(groups[card.ChanKey], task)
		}
		for chanKey, tasks := range groups {
			stage = "enqueue"
			if err = lockCtx.Err(); err != nil {
				return err
			}
			added, enqueueErr := workStore.EnqueueRunBatch(lockCtx, message.RunID, chanKey, tasks, time.Now())
			if enqueueErr != nil {
				return fmt.Errorf("enqueue imei channel %s: %w", chanKey, enqueueErr)
			}
			stats.enqueued += int64(added)
		}
		stage = "capacity"
		if err = p.waitForCapacity(lockCtx); err != nil {
			return err
		}
	}
	stage = "producer_done"
	if err = lockCtx.Err(); err != nil {
		return err
	}
	return workStore.MarkProducerDone(lockCtx, message.RunID)
}

// scopeDigest is a stable short fingerprint of the selected channels.
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])
}

func (p *Producer) renewLock(ctx context.Context, cancel context.CancelCauseFunc, leaseStore *queue.Store) {
	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:
			if ctx.Err() != nil {
				return
			}
			renewCtx, stop := context.WithTimeout(ctx, leaseIOBudget(p.config.RunLockTTL))
			ok, err := leaseStore.RenewLock(renewCtx, p.config.RunLockKey, p.owner, p.config.RunLockTTL)
			stop()
			if ctx.Err() != nil {
				return
			}
			if err == nil && !ok {
				err = ErrAlreadyRunning
			}
			if err != nil {
				cancel(err)
				return
			}
		}
	}
}

func (p *Producer) waitForCapacity(ctx context.Context) error {
	if ctx.Err() != nil {
		return context.Cause(ctx)
	}
	if p.config.GlobalHighWaterMark <= 0 {
		return nil
	}
	for {
		if ctx.Err() != nil {
			return context.Cause(ctx)
		}
		runs, err := p.store.ActiveRuns(ctx)
		if err != nil {
			return err
		}
		var total int64
		for _, runID := range runs {
			meta, sizeErr := p.store.GetRunMeta(ctx, runID)
			if sizeErr != nil {
				return sizeErr
			}
			// ActiveCount includes leased tasks; do not count InflightCount twice.
			total += meta.ActiveCount + meta.NextRoundCount
		}
		if total <= p.config.GlobalHighWaterMark {
			return nil
		}
		select {
		case <-ctx.Done():
			return context.Cause(ctx)
		case <-time.After(p.config.PollInterval):
		}
	}
}

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