hyapp-server/services/activity-service/internal/service/activitytemplate/runtime.go

package activitytemplate

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"log/slog"
	"net/url"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/protobuf/proto"
	roomeventsv1 "hyapp.local/api/proto/events/room/v1"
	walletv1 "hyapp.local/api/proto/wallet/v1"
	"hyapp/pkg/activitymq"
	"hyapp/pkg/appcode"
	"hyapp/pkg/idgen"
	"hyapp/pkg/logx"
	"hyapp/pkg/rocketmqx"
	"hyapp/pkg/xerr"
	domain "hyapp/services/activity-service/internal/domain/activitytemplate"
	messageservice "hyapp/services/activity-service/internal/service/message"
)

const (
	activityTemplateTaskRewardReason = "activity template daily task reward"
	activityTemplateRankRewardReason = "activity template leaderboard reward"
	activityTemplateTaskGrantSource  = "activity_template_task"
	activityTemplateRankGrantSource  = "activity_template_rank"
)

// RuntimeRepository is intentionally separate from the configuration Repository so legacy unit-test fakes
// remain small. Production MySQL implements both and New binds them through a checked type assertion.
type RuntimeRepository interface {
	FindCurrentActivityTemplateRuntime(ctx context.Context, templateCode string, regionID int64, atMS int64) (domain.Runtime, bool, error)
	FindActivityTemplateRuntimeForPeriod(ctx context.Context, templateCode string, versionNo int64, regionID int64, boardType string, periodKey string, atMS int64) (domain.Runtime, bool, error)
	FindActivityTemplateRuntimeForUserTask(ctx context.Context, templateCode string, versionNo int64, userID int64, taskDay string, taskKey string) (domain.Runtime, bool, error)
	FindActivityTemplateRuntimeForExactLeaderboard(ctx context.Context, templateCode string, versionNo int64, boardType string, periodKey string, atMS int64) (domain.Runtime, bool, error)
	GetActivityTemplateRuntimeVersion(ctx context.Context, templateID string, versionNo int64) (domain.Runtime, error)
	ConsumeActivityTemplateGiftEvent(ctx context.Context, event domain.GiftEvent, nowMS int64) (domain.EventResult, error)
	VisitActivityTemplate(ctx context.Context, command domain.VisitCommand, nowMS int64) (domain.VisitResult, error)
	ListActivityTemplateTaskProgress(ctx context.Context, runtime domain.Runtime, userID int64, taskDay string) ([]domain.TaskProgress, error)
	PrepareActivityTemplateTaskClaim(ctx context.Context, runtime domain.Runtime, command domain.ClaimCommand, nowMS int64) (domain.TaskClaim, error)
	MarkActivityTemplateTaskClaimGranted(ctx context.Context, claimID string, walletGrantID string, nowMS int64) (domain.TaskClaim, error)
	MarkActivityTemplateTaskClaimFailed(ctx context.Context, claimID string, reason string, nowMS int64) error
	ListActivityTemplateRuntimeLeaderboard(ctx context.Context, runtime domain.Runtime, query domain.LeaderboardQuery) (domain.LeaderboardResult, error)
	ClaimDueActivityTemplateRankPeriods(ctx context.Context, workerID string, nowMS int64, lockTTL time.Duration, batchSize int32) ([]domain.RankPeriod, error)
	PrepareActivityTemplateRankSettlement(ctx context.Context, period domain.RankPeriod, nowMS int64) ([]domain.RewardJob, error)
	ClaimFinalizableActivityTemplateRankPeriods(ctx context.Context, workerID string, nowMS int64, lockTTL time.Duration, batchSize int32) ([]domain.RankPeriod, error)
	ClaimActivityTemplateRewardJobs(ctx context.Context, workerID string, nowMS int64, lockTTL time.Duration, batchSize int32) ([]domain.RewardJob, error)
	ExtendActivityTemplateRewardJobLease(ctx context.Context, rewardJobID string, workerID string, nowMS int64, lockTTL time.Duration) error
	MarkActivityTemplateRewardJobGrantedWithLease(ctx context.Context, rewardJobID string, workerID string, walletGrantID string, nowMS int64) error
	MarkActivityTemplateRewardJobFailedWithLease(ctx context.Context, rewardJobID string, workerID string, reason string, nowMS int64) error
	FinishActivityTemplateRankPeriodIfComplete(ctx context.Context, period domain.RankPeriod, nowMS int64) (bool, error)
	ListActivityTemplateRewardJobs(ctx context.Context, query domain.RewardJobQuery) ([]domain.RewardJob, int64, error)
	ClaimActivityTemplateRewardJobForManualRetry(ctx context.Context, templateID string, rewardJobID string, workerID string, nowMS int64, lockTTL time.Duration) (domain.RewardJob, error)
	ListActivityTemplateTaskClaims(ctx context.Context, query domain.TaskClaimQuery) ([]domain.TaskClaim, int64, error)
	ClaimActivityTemplateTaskClaimForRetry(ctx context.Context, templateID string, claimID string, nowMS int64) (domain.TaskClaim, error)
	ClaimActivityTemplateStatsOutbox(ctx context.Context, workerID string, nowMS int64, lockTTL time.Duration, batchSize int) ([]domain.StatsOutboxRecord, error)
	MarkActivityTemplateStatsOutboxSent(ctx context.Context, outboxID string, nowMS int64) error
	MarkActivityTemplateStatsOutboxFailed(ctx context.Context, outboxID string, reason string, nowMS int64) error
}

type RuntimeWallet interface {
	GrantPinnedResourceGroup(ctx context.Context, req *walletv1.GrantPinnedResourceGroupRequest, opts ...grpc.CallOption) (*walletv1.ResourceGrantResponse, error)
}

type StatsPublisher interface {
	SendSync(ctx context.Context, message rocketmqx.Message) error
}

type NoticeService interface {
	CreateActivityNotice(ctx context.Context, cmd messageservice.NoticeCommand) (messageservice.NoticeResult, error)
}

// BindRuntime installs optional runtime dependencies without widening the stable configuration constructor.
// Production wiring treats a missing binding as Unavailable; configuration-only tests never call these methods.
func (s *Service) BindRuntime(repository RuntimeRepository, wallet RuntimeWallet) {
	if s == nil {
		return
	}
	s.runtime = repository
	s.runtimeWallet = wallet
}

func (s *Service) BindNotice(notices NoticeService) {
	if s != nil {
		s.notices = notices
	}
}

func (s *Service) GetCurrentRuntime(ctx context.Context, regionID int64, templateCode string) (domain.Runtime, bool, error) {
	if err := s.requireRuntime(); err != nil {
		return domain.Runtime{}, false, err
	}
	if regionID <= 0 {
		return domain.Runtime{}, false, xerr.New(xerr.InvalidArgument, "region_id is required")
	}
	return s.runtime.FindCurrentActivityTemplateRuntime(ctx, strings.TrimSpace(templateCode), regionID, s.now().UTC().UnixMilli())
}

func (s *Service) Visit(ctx context.Context, command domain.VisitCommand) (domain.VisitResult, error) {
	if err := s.requireRuntime(); err != nil {
		return domain.VisitResult{}, err
	}
	command.TemplateCode = strings.TrimSpace(command.TemplateCode)
	command.CommandID = strings.TrimSpace(command.CommandID)
	if command.UserID <= 0 || command.RegionID <= 0 || command.VersionNo <= 0 || command.TemplateCode == "" || command.CommandID == "" || len(command.CommandID) > 128 {
		return domain.VisitResult{}, xerr.New(xerr.InvalidArgument, "visit command is incomplete")
	}
	return s.runtime.VisitActivityTemplate(ctx, command, s.now().UTC().UnixMilli())
}

func (s *Service) ListRuntimeTasks(ctx context.Context, userID, regionID, versionNo int64, templateCode, taskDay string) (domain.Runtime, []domain.TaskProgress, int64, error) {
	if err := s.requireRuntime(); err != nil {
		return domain.Runtime{}, nil, 0, err
	}
	if userID <= 0 || versionNo < 0 || strings.TrimSpace(templateCode) == "" || (versionNo == 0 && regionID <= 0) {
		return domain.Runtime{}, nil, 0, xerr.New(xerr.InvalidArgument, "task query is incomplete")
	}
	nowMS := s.now().UTC().UnixMilli()
	var runtime domain.Runtime
	var found bool
	var err error
	if versionNo > 0 {
		if strings.TrimSpace(taskDay) == "" {
			return domain.Runtime{}, nil, 0, xerr.New(xerr.InvalidArgument, "task_day is required for historical version")
		}
		runtime, found, err = s.runtime.FindActivityTemplateRuntimeForUserTask(ctx, strings.TrimSpace(templateCode), versionNo, userID, taskDay, "")
	} else {
		runtime, found, err = s.runtime.FindCurrentActivityTemplateRuntime(ctx, strings.TrimSpace(templateCode), regionID, nowMS)
	}
	if err != nil {
		return domain.Runtime{}, nil, 0, err
	}
	if !found {
		return domain.Runtime{}, nil, 0, xerr.New(xerr.NotFound, "active activity template not found")
	}
	if strings.TrimSpace(taskDay) == "" {
		taskDay = utcDay(nowMS)
	}
	dayStart, err := parseUTCDay(taskDay)
	if err != nil || dayStart >= runtime.Template.EndMS || dayStart+int64(24*time.Hour/time.Millisecond) <= runtime.RuntimeFromMS {
		return domain.Runtime{}, nil, 0, xerr.New(xerr.InvalidArgument, "task_day is outside the activity runtime")
	}
	items, err := s.runtime.ListActivityTemplateTaskProgress(ctx, runtime, userID, taskDay)
	if err != nil {
		return domain.Runtime{}, nil, 0, err
	}
	nextRefresh := dayStart + int64(24*time.Hour/time.Millisecond)
	if runtime.Template.EndMS < nextRefresh {
		nextRefresh = runtime.Template.EndMS
	}
	return runtime, items, nextRefresh, nil
}

func (s *Service) ClaimRuntimeTaskReward(ctx context.Context, command domain.ClaimCommand) (domain.TaskClaim, error) {
	if err := s.requireRuntime(); err != nil {
		return domain.TaskClaim{}, err
	}
	if s.runtimeWallet == nil {
		return domain.TaskClaim{}, xerr.New(xerr.Unavailable, "wallet client is not configured")
	}
	command.TemplateCode = strings.TrimSpace(command.TemplateCode)
	command.TaskDay = strings.TrimSpace(command.TaskDay)
	command.TaskKey = strings.TrimSpace(command.TaskKey)
	command.CommandID = strings.TrimSpace(command.CommandID)
	if command.UserID <= 0 || command.TemplateCode == "" || command.VersionNo <= 0 || command.TaskKey == "" || command.CommandID == "" || len(command.CommandID) > 128 {
		return domain.TaskClaim{}, xerr.New(xerr.InvalidArgument, "task claim command is incomplete")
	}
	nowMS := s.now().UTC().UnixMilli()
	if command.TaskDay == "" {
		command.TaskDay = utcDay(nowMS)
	}
	if _, err := parseUTCDay(command.TaskDay); err != nil {
		return domain.TaskClaim{}, err
	}
	// version_no 来自任务列表的不可变版本。即使同一 template_code 在同一个 UTC 日重发，v1 的失败
	// claim 也不会被 latest-v2 吞掉；物化 period 仍允许活动结束后的幂等 wallet 补偿。
	runtime, found, err := s.runtime.FindActivityTemplateRuntimeForUserTask(
		ctx, command.TemplateCode, command.VersionNo, command.UserID, command.TaskDay, command.TaskKey,
	)
	if err != nil {
		return domain.TaskClaim{}, err
	}
	if !found {
		return domain.TaskClaim{}, xerr.New(xerr.NotFound, "activity template task period not found")
	}
	claim, err := s.runtime.PrepareActivityTemplateTaskClaim(ctx, runtime, command, nowMS)
	if err != nil || claim.Status == domain.ClaimStatusGranted {
		return claim, err
	}
	if claim.RewardSnapshot.SnapshotID == "" {
		err := xerr.New(xerr.Conflict, "activity task reward snapshot is missing")
		_ = s.runtime.MarkActivityTemplateTaskClaimFailed(ctx, claim.ClaimID, xerr.MessageOf(err), s.now().UTC().UnixMilli())
		return domain.TaskClaim{}, err
	}
	resp, err := s.runtimeWallet.GrantPinnedResourceGroup(ctx, &walletv1.GrantPinnedResourceGroupRequest{
		CommandId: claim.WalletCommandID, AppCode: appcode.FromContext(ctx), TargetUserId: claim.UserID,
		SnapshotId: claim.RewardSnapshot.SnapshotID, Reason: activityTemplateTaskRewardReason,
		OperatorUserId: claim.UserID, GrantSource: activityTemplateTaskGrantSource,
		ExpectedSnapshotHash: claim.RewardSnapshot.SnapshotHash,
	})
	if err != nil {
		_ = s.runtime.MarkActivityTemplateTaskClaimFailed(ctx, claim.ClaimID, xerr.MessageOf(err), s.now().UTC().UnixMilli())
		return domain.TaskClaim{}, err
	}
	walletGrantID := strings.TrimSpace(resp.GetGrant().GetGrantId())
	if walletGrantID == "" {
		err := xerr.New(xerr.Internal, "wallet returned an empty resource grant")
		_ = s.runtime.MarkActivityTemplateTaskClaimFailed(ctx, claim.ClaimID, xerr.MessageOf(err), s.now().UTC().UnixMilli())
		return domain.TaskClaim{}, err
	}
	return s.runtime.MarkActivityTemplateTaskClaimGranted(ctx, claim.ClaimID, walletGrantID, s.now().UTC().UnixMilli())
}

func (s *Service) ListRuntimeLeaderboard(ctx context.Context, query domain.LeaderboardQuery) (domain.LeaderboardResult, error) {
	if err := s.requireRuntime(); err != nil {
		return domain.LeaderboardResult{}, err
	}
	query.TemplateCode = strings.TrimSpace(query.TemplateCode)
	query.BoardType = strings.ToLower(strings.TrimSpace(query.BoardType))
	query.PeriodKey = strings.TrimSpace(query.PeriodKey)
	if query.UserID <= 0 || query.VersionNo < 0 || query.TemplateCode == "" || (query.VersionNo == 0 && query.RegionID <= 0) || (query.BoardType != domain.BoardTypeDaily && query.BoardType != domain.BoardTypeTotal) {
		return domain.LeaderboardResult{}, xerr.New(xerr.InvalidArgument, "leaderboard query is invalid")
	}
	query.NowMS = s.now().UTC().UnixMilli()
	var runtime domain.Runtime
	var found bool
	var err error
	// 显式 period_key 是可分享、可从获奖通知回跳的稳定榜单身份。先按物化 period 解析其不可变发布版本，
	// 因此活动结束或同一模版中途重发后仍能读取冻结榜；未显式指定时继续只暴露当前运行版本。
	if query.VersionNo > 0 && query.PeriodKey == "" {
		if query.BoardType == domain.BoardTypeTotal {
			query.PeriodKey = "total"
		} else {
			query.PeriodKey = utcDay(query.NowMS)
		}
	}
	if query.PeriodKey != "" && query.VersionNo > 0 {
		runtime, found, err = s.runtime.FindActivityTemplateRuntimeForExactLeaderboard(
			ctx, query.TemplateCode, query.VersionNo, query.BoardType, query.PeriodKey, query.NowMS,
		)
	} else if query.PeriodKey != "" {
		runtime, found, err = s.runtime.FindActivityTemplateRuntimeForPeriod(
			ctx, query.TemplateCode, query.VersionNo, query.RegionID, query.BoardType, query.PeriodKey, query.NowMS,
		)
	} else {
		runtime, found, err = s.runtime.FindCurrentActivityTemplateRuntime(ctx, query.TemplateCode, query.RegionID, query.NowMS)
	}
	if err != nil {
		return domain.LeaderboardResult{}, err
	}
	if !found {
		return domain.LeaderboardResult{}, xerr.New(xerr.NotFound, "active activity template not found")
	}
	return s.runtime.ListActivityTemplateRuntimeLeaderboard(ctx, runtime, query)
}

func (s *Service) AdminListRuntimeLeaderboard(ctx context.Context, query domain.LeaderboardQuery) (domain.LeaderboardResult, error) {
	if err := s.requireRuntime(); err != nil {
		return domain.LeaderboardResult{}, err
	}
	query.TemplateID = strings.TrimSpace(query.TemplateID)
	query.BoardType = strings.ToLower(strings.TrimSpace(query.BoardType))
	if query.TemplateID == "" || query.VersionNo <= 0 || (query.BoardType != domain.BoardTypeDaily && query.BoardType != domain.BoardTypeTotal) {
		return domain.LeaderboardResult{}, xerr.New(xerr.InvalidArgument, "admin leaderboard query is invalid")
	}
	query.NowMS = s.now().UTC().UnixMilli()
	runtime, err := s.runtime.GetActivityTemplateRuntimeVersion(ctx, query.TemplateID, query.VersionNo)
	if err != nil {
		return domain.LeaderboardResult{}, err
	}
	return s.runtime.ListActivityTemplateRuntimeLeaderboard(ctx, runtime, query)
}

func (s *Service) ListRewardJobs(ctx context.Context, query domain.RewardJobQuery) ([]domain.RewardJob, int64, error) {
	if err := s.requireRuntime(); err != nil {
		return nil, 0, err
	}
	query.TemplateID = strings.TrimSpace(query.TemplateID)
	query.BoardType = strings.ToLower(strings.TrimSpace(query.BoardType))
	query.PeriodKey = strings.TrimSpace(query.PeriodKey)
	query.Status = strings.ToLower(strings.TrimSpace(query.Status))
	if query.TemplateID == "" {
		return nil, 0, xerr.New(xerr.InvalidArgument, "template_id is required")
	}
	if query.BoardType != "" && query.BoardType != domain.BoardTypeDaily && query.BoardType != domain.BoardTypeTotal {
		return nil, 0, xerr.New(xerr.InvalidArgument, "board_type is invalid")
	}
	if query.Status != "" && query.Status != domain.ClaimStatusPending && query.Status != domain.ClaimStatusRunning && query.Status != domain.ClaimStatusGranted && query.Status != domain.ClaimStatusFailed && query.Status != domain.ClaimStatusDead {
		return nil, 0, xerr.New(xerr.InvalidArgument, "reward status is invalid")
	}
	return s.runtime.ListActivityTemplateRewardJobs(ctx, query)
}

func (s *Service) ListTaskClaims(ctx context.Context, query domain.TaskClaimQuery) ([]domain.TaskClaim, int64, error) {
	if err := s.requireRuntime(); err != nil {
		return nil, 0, err
	}
	query.TemplateID = strings.TrimSpace(query.TemplateID)
	query.TaskDay = strings.TrimSpace(query.TaskDay)
	query.TaskKey = strings.TrimSpace(query.TaskKey)
	query.Status = strings.ToLower(strings.TrimSpace(query.Status))
	if query.TemplateID == "" || query.VersionNo < 0 || query.UserID < 0 {
		return nil, 0, xerr.New(xerr.InvalidArgument, "task claim query is invalid")
	}
	if query.TaskDay != "" {
		if _, err := parseUTCDay(query.TaskDay); err != nil {
			return nil, 0, err
		}
	}
	if query.Status != "" && query.Status != domain.ClaimStatusPending && query.Status != domain.ClaimStatusRunning && query.Status != domain.ClaimStatusGranted && query.Status != domain.ClaimStatusFailed {
		return nil, 0, xerr.New(xerr.InvalidArgument, "task claim status is invalid")
	}
	return s.runtime.ListActivityTemplateTaskClaims(ctx, query)
}

func (s *Service) RetryTaskClaim(ctx context.Context, templateID, claimID string, operatorAdminID int64) (domain.TaskClaim, error) {
	if err := s.requireRuntime(); err != nil {
		return domain.TaskClaim{}, err
	}
	if s.runtimeWallet == nil {
		return domain.TaskClaim{}, xerr.New(xerr.Unavailable, "wallet client is not configured")
	}
	if strings.TrimSpace(templateID) == "" || strings.TrimSpace(claimID) == "" || operatorAdminID <= 0 {
		return domain.TaskClaim{}, xerr.New(xerr.InvalidArgument, "task claim retry command is incomplete")
	}
	claim, err := s.runtime.ClaimActivityTemplateTaskClaimForRetry(ctx, strings.TrimSpace(templateID), strings.TrimSpace(claimID), s.now().UTC().UnixMilli())
	if err != nil {
		return domain.TaskClaim{}, err
	}
	if claim.RewardSnapshot.SnapshotID == "" || !validSHA256Hex(claim.RewardSnapshot.SnapshotHash) {
		err := xerr.New(xerr.Conflict, "activity task reward snapshot is missing")
		_ = s.runtime.MarkActivityTemplateTaskClaimFailed(ctx, claim.ClaimID, xerr.MessageOf(err), s.now().UTC().UnixMilli())
		return domain.TaskClaim{}, err
	}
	resp, err := s.runtimeWallet.GrantPinnedResourceGroup(ctx, &walletv1.GrantPinnedResourceGroupRequest{
		CommandId: claim.WalletCommandID, AppCode: appcode.FromContext(ctx), TargetUserId: claim.UserID,
		SnapshotId: claim.RewardSnapshot.SnapshotID, ExpectedSnapshotHash: claim.RewardSnapshot.SnapshotHash,
		Reason: activityTemplateTaskRewardReason, OperatorUserId: claim.UserID, GrantSource: activityTemplateTaskGrantSource,
	})
	if err != nil {
		_ = s.runtime.MarkActivityTemplateTaskClaimFailed(ctx, claim.ClaimID, xerr.MessageOf(err), s.now().UTC().UnixMilli())
		return domain.TaskClaim{}, err
	}
	walletGrantID := strings.TrimSpace(resp.GetGrant().GetGrantId())
	if walletGrantID == "" {
		err := xerr.New(xerr.Internal, "wallet returned an empty resource grant")
		_ = s.runtime.MarkActivityTemplateTaskClaimFailed(ctx, claim.ClaimID, xerr.MessageOf(err), s.now().UTC().UnixMilli())
		return domain.TaskClaim{}, err
	}
	return s.runtime.MarkActivityTemplateTaskClaimGranted(ctx, claim.ClaimID, walletGrantID, s.now().UTC().UnixMilli())
}

// ProcessRuntimeSettlementBatch separates bounded period materialization from bounded job delivery. A large
// leaderboard can therefore never make one cron RPC perform unbounded wallet calls; every delivery is fenced by
// a per-invocation lease owner and remains replay-safe through the immutable wallet command.
func (s *Service) ProcessRuntimeSettlementBatch(ctx context.Context, runID, workerID string, batchSize int32, lockTTL time.Duration) (claimed, processed, success, failure int32, hasMore bool, err error) {
	if err := s.requireRuntime(); err != nil {
		return 0, 0, 0, 0, false, err
	}
	if s.runtimeWallet == nil {
		return 0, 0, 0, 0, false, xerr.New(xerr.Unavailable, "wallet client is not configured")
	}
	if batchSize <= 0 {
		batchSize = 20
	}
	nowMS := s.now().UTC().UnixMilli()
	workerID = strings.TrimSpace(workerID)
	if workerID == "" {
		workerID = "activity-template-settlement"
	}
	runID = strings.TrimSpace(runID)
	if runID == "" {
		runID = idgen.New("run")
	}
	leaseOwner := activityTemplateRewardLeaseOwner(workerID, runID)
	periods, err := s.runtime.ClaimDueActivityTemplateRankPeriods(ctx, leaseOwner, nowMS, lockTTL, batchSize)
	if err != nil {
		return 0, 0, 0, 0, false, err
	}
	for _, period := range periods {
		_, prepareErr := s.runtime.PrepareActivityTemplateRankSettlement(ctx, period, s.now().UTC().UnixMilli())
		if prepareErr != nil {
			logx.Error(ctx, "activity_template_prepare_settlement_failed", prepareErr, slog.String("template_id", period.TemplateID), slog.String("period_key", period.PeriodKey), slog.String("run_id", runID))
			failure++
			continue
		}
		// Empty/no-reward boards settle here; boards with jobs are released from the period lock and
		// finish after the independent job batch reaches all-granted.
		if _, finishErr := s.runtime.FinishActivityTemplateRankPeriodIfComplete(ctx, period, s.now().UTC().UnixMilli()); finishErr != nil {
			failure++
		}
	}

	jobs, err := s.runtime.ClaimActivityTemplateRewardJobs(ctx, leaseOwner, s.now().UTC().UnixMilli(), lockTTL, batchSize)
	if err != nil {
		return 0, 0, 0, failure, len(periods) == int(batchSize), err
	}
	claimed = int32(len(jobs))
	processed = int32(len(jobs))
	var succeeded atomic.Int32
	var failed atomic.Int32
	workerCount := min(len(jobs), 8)
	if workerCount > 0 {
		queue := make(chan domain.RewardJob)
		var workers sync.WaitGroup
		workers.Add(workerCount)
		for range workerCount {
			go func() {
				defer workers.Done()
				for job := range queue {
					if _, grantErr := s.grantRuntimeRewardJob(ctx, job, leaseOwner, lockTTL); grantErr != nil {
						failed.Add(1)
						continue
					}
					succeeded.Add(1)
				}
			}()
		}
		for _, job := range jobs {
			queue <- job
		}
		close(queue)
		workers.Wait()
	}
	success = succeeded.Load()
	failure += failed.Load()

	// A single job batch can span periods. Finalize each identity once after every wallet worker has stopped,
	// so a period is never marked settled while an in-process job still owns a lease.
	periodsToFinish := make(map[string]domain.RankPeriod)
	for _, job := range jobs {
		key := job.TemplateID + "\x00" + strconv.FormatInt(job.VersionNo, 10) + "\x00" + job.BoardType + "\x00" + job.PeriodKey
		periodsToFinish[key] = domain.RankPeriod{
			AppCode: job.AppCode, TemplateID: job.TemplateID, TemplateCode: job.TemplateCode, VersionNo: job.VersionNo,
			BoardType: job.BoardType, PeriodKey: job.PeriodKey,
		}
	}
	for _, period := range periodsToFinish {
		if _, finishErr := s.runtime.FinishActivityTemplateRankPeriodIfComplete(ctx, period, s.now().UTC().UnixMilli()); finishErr != nil {
			failure++
		}
	}

	// Recover the narrow crash window after the last granted-job commit but before its period transition. The
	// repository only returns `rewarding` periods with zero non-granted jobs, so failed/dead/backoff queues cannot
	// consume this bound or keep cron in a has_more hot loop.
	finalizable, finalizeClaimErr := s.runtime.ClaimFinalizableActivityTemplateRankPeriods(
		ctx, leaseOwner, s.now().UTC().UnixMilli(), lockTTL, batchSize,
	)
	if finalizeClaimErr != nil {
		return claimed, processed, success, failure, len(periods) == int(batchSize) || len(jobs) == int(batchSize), finalizeClaimErr
	}
	for _, period := range finalizable {
		if _, finishErr := s.runtime.FinishActivityTemplateRankPeriodIfComplete(ctx, period, s.now().UTC().UnixMilli()); finishErr != nil {
			failure++
		}
	}
	hasMore = len(periods) == int(batchSize) || len(jobs) == int(batchSize) || len(finalizable) == int(batchSize)
	return claimed, processed, success, failure, hasMore, nil
}

func (s *Service) RetryRewardJob(ctx context.Context, templateID, rewardJobID string, operatorAdminID int64) (domain.RewardJob, error) {
	if err := s.requireRuntime(); err != nil {
		return domain.RewardJob{}, err
	}
	if s.runtimeWallet == nil {
		return domain.RewardJob{}, xerr.New(xerr.Unavailable, "wallet client is not configured")
	}
	if strings.TrimSpace(templateID) == "" || strings.TrimSpace(rewardJobID) == "" || operatorAdminID <= 0 {
		return domain.RewardJob{}, xerr.New(xerr.InvalidArgument, "reward retry command is incomplete")
	}
	nowMS := s.now().UTC().UnixMilli()
	leaseOwner := activityTemplateRewardLeaseOwner("admin-retry-"+strconv.FormatInt(operatorAdminID, 10), idgen.New("request")+":"+rewardJobID)
	const manualRetryLeaseTTL = time.Minute
	job, err := s.runtime.ClaimActivityTemplateRewardJobForManualRetry(
		ctx, strings.TrimSpace(templateID), strings.TrimSpace(rewardJobID), leaseOwner, nowMS, manualRetryLeaseTTL,
	)
	if err != nil || job.Status == domain.ClaimStatusGranted {
		return job, err
	}
	walletGrantID, err := s.grantRuntimeRewardJob(ctx, job, leaseOwner, manualRetryLeaseTTL)
	if err != nil {
		return domain.RewardJob{}, err
	}
	job.Status = domain.ClaimStatusGranted
	job.WalletGrantID = walletGrantID
	job.FailureReason = ""
	job.GrantedAtMS = s.now().UTC().UnixMilli()
	job.UpdatedAtMS = job.GrantedAtMS
	return job, nil
}

func (s *Service) grantRuntimeRewardJob(ctx context.Context, job domain.RewardJob, leaseOwner string, lockTTL time.Duration) (string, error) {
	markFailed := func(cause error) error {
		if markErr := s.runtime.MarkActivityTemplateRewardJobFailedWithLease(ctx, job.RewardJobID, leaseOwner, xerr.MessageOf(cause), s.now().UTC().UnixMilli()); markErr != nil {
			logx.Error(ctx, "activity_template_reward_job_failure_fence_lost", markErr,
				slog.String("reward_job_id", job.RewardJobID), slog.String("lease_owner", leaseOwner))
		}
		return cause
	}
	if job.RewardSnapshot.SnapshotID == "" || !validSHA256Hex(job.RewardSnapshot.SnapshotHash) {
		err := xerr.New(xerr.Conflict, "activity rank reward snapshot is missing")
		return "", markFailed(err)
	}
	if err := s.runtime.ExtendActivityTemplateRewardJobLease(ctx, job.RewardJobID, leaseOwner, s.now().UTC().UnixMilli(), lockTTL); err != nil {
		return "", err
	}
	resp, err := s.runtimeWallet.GrantPinnedResourceGroup(ctx, &walletv1.GrantPinnedResourceGroupRequest{
		CommandId: job.WalletCommandID, AppCode: appcode.FromContext(ctx), TargetUserId: job.UserID,
		SnapshotId: job.RewardSnapshot.SnapshotID, Reason: activityTemplateRankRewardReason,
		OperatorUserId: job.UserID, GrantSource: activityTemplateRankGrantSource,
		ExpectedSnapshotHash: job.RewardSnapshot.SnapshotHash,
	})
	if err != nil {
		return "", markFailed(err)
	}
	walletGrantID := strings.TrimSpace(resp.GetGrant().GetGrantId())
	if walletGrantID == "" {
		err := xerr.New(xerr.Internal, "wallet returned an empty resource grant")
		return "", markFailed(err)
	}
	// The wallet call can outlive the original lease window. Re-fence before emitting the user notice; if
	// another worker took over, it will replay the same wallet command and stable notice event instead.
	if err := s.runtime.ExtendActivityTemplateRewardJobLease(ctx, job.RewardJobID, leaseOwner, s.now().UTC().UnixMilli(), lockTTL); err != nil {
		return "", err
	}
	if s.notices != nil {
		action := url.Values{}
		action.Set("template_code", job.TemplateCode)
		action.Set("version_no", strconv.FormatInt(job.VersionNo, 10))
		action.Set("board_type", job.BoardType)
		action.Set("period_key", job.PeriodKey)
		_, noticeErr := s.notices.CreateActivityNotice(ctx, messageservice.NoticeCommand{
			TargetUserID: job.UserID, Producer: "activity-template-runtime",
			ProducerEventID: job.RewardJobID + ":granted", ProducerEventType: "ActivityTemplateRankRewardGranted",
			AggregateType: "activity_template_reward_job", AggregateID: job.RewardJobID,
			TemplateID: job.TemplateID, TemplateVersion: strconv.FormatInt(job.VersionNo, 10),
			// Inbox storage currently requires title/summary text. Persist stable content keys rather than
			// English prose; App clients localize the keys with the immutable metadata parameters below.
			Title: "activity_template.rank_reward.title", Summary: "activity_template.rank_reward.summary",
			ActionType:  "activity_detail",
			ActionParam: action.Encode(), SentAtMS: s.now().UTC().UnixMilli(),
			Metadata: map[string]any{
				"content_key": "activity_template.rank_reward_granted", "template_code": job.TemplateCode,
				"version_no": job.VersionNo, "board_type": job.BoardType, "period_key": job.PeriodKey,
				"rank_no": job.RankNo, "reward_name": job.RewardSnapshot.Name,
				"resource_group_id": job.ResourceGroupID,
			},
		})
		if noticeErr != nil {
			// Wallet command is stable. Marking failed makes the next cron/admin retry replay the same wallet
			// grant (idempotent) and retry the inbox write instead of silently losing the winner notification.
			return "", markFailed(noticeErr)
		}
	}
	if err := s.runtime.MarkActivityTemplateRewardJobGrantedWithLease(ctx, job.RewardJobID, leaseOwner, walletGrantID, s.now().UTC().UnixMilli()); err != nil {
		// 钱包与站内信均已幂等成功但本地状态落库失败时，立即释放结算周期并标为 failed；下一轮会用同一
		// wallet_command_id/producer_event_id 补偿，不会重复发奖或重复通知。
		_ = s.runtime.MarkActivityTemplateRewardJobFailedWithLease(ctx, job.RewardJobID, leaseOwner, xerr.MessageOf(err), s.now().UTC().UnixMilli())
		return "", err
	}
	return walletGrantID, nil
}

func activityTemplateRewardLeaseOwner(workerID, invocationID string) string {
	workerID = strings.TrimSpace(workerID)
	invocationID = strings.TrimSpace(invocationID)
	owner := workerID + ":" + invocationID
	if len(owner) <= 128 {
		return owner
	}
	// Preserve an operator-readable prefix while bounding the persisted fencing token.
	digest := sha256.Sum256([]byte(owner))
	return "activity-template-reward-lease:" + hex.EncodeToString(digest[:])
}

// HandleRoomEvent accepts only wallet-committed real gift facts. coin_spent is the scoring source of truth;
// configured gift price is a display snapshot and is never multiplied in this path.
func (s *Service) HandleRoomEvent(ctx context.Context, envelope *roomeventsv1.EventEnvelope) (int32, error) {
	if s == nil || s.runtime == nil || envelope == nil || envelope.GetEventType() != "RoomGiftSent" {
		return 0, nil
	}
	var gift roomeventsv1.RoomGiftSent
	if err := proto.Unmarshal(envelope.GetBody(), &gift); err != nil {
		return 0, err
	}
	if gift.GetIsRobotGift() || gift.GetSyntheticLuckyGift() || gift.GetRealRoomRobotGift() {
		return 0, nil
	}
	if strings.TrimSpace(envelope.GetEventId()) == "" || gift.GetSenderUserId() <= 0 || strings.TrimSpace(gift.GetGiftId()) == "" || gift.GetGiftCount() <= 0 || gift.GetCoinSpent() <= 0 || gift.GetRegionId() <= 0 || envelope.GetOccurredAtMs() <= 0 {
		return 0, nil
	}
	result, err := s.runtime.ConsumeActivityTemplateGiftEvent(appcode.WithContext(ctx, envelope.GetAppCode()), domain.GiftEvent{
		EventID: strings.TrimSpace(envelope.GetEventId()), UserID: gift.GetSenderUserId(), GiftID: strings.TrimSpace(gift.GetGiftId()),
		// region_id is the sender's profile region captured by room-service at wallet commit. visible_region_id
		// describes the room audience and must never choose a user-targeted activity runtime.
		GiftCount: int64(gift.GetGiftCount()), CoinAmount: gift.GetCoinSpent(), RegionID: gift.GetRegionId(),
		OccurredAtMS: envelope.GetOccurredAtMs(),
	}, s.now().UTC().UnixMilli())
	if err != nil {
		return 0, err
	}
	// Normal misses (no active runtime / gift not configured) are expected routing outcomes. Closed rank
	// watermarks are operationally different: a committed gift arrived too late for at least one projection,
	// so surface both partial acceptance and full rejection with delay metadata for alerting.
	if strings.Contains(result.SkipReason, "settlement_window_closed") || strings.Contains(result.SkipReason, "rank_period_closed") {
		logx.Warn(ctx, "activity_template_gift_rank_projection_late",
			slog.String("event_id", result.EventID), slog.String("template_id", result.TemplateID),
			slog.Int64("version_no", result.VersionNo), slog.String("task_day", result.TaskDay),
			slog.String("projection_outcome", result.SkipReason), slog.Int64("processing_delay_ms", max(int64(0), s.now().UTC().UnixMilli()-envelope.GetOccurredAtMs())))
	}
	if result.Status == domain.EventStatusConsumed {
		return 1, nil
	}
	return 0, nil
}

// ProcessStatsOutboxBatch publishes owner facts to the statistics topic. Sent is marked only after broker ack;
// a crash after ack but before MarkSent is harmless because statistics_event_consumption deduplicates event_id.
func (s *Service) ProcessStatsOutboxBatch(ctx context.Context, publisher StatsPublisher, topic, workerID string, batchSize int, lockTTL time.Duration) (int, int, error) {
	if err := s.requireRuntime(); err != nil {
		return 0, 0, err
	}
	if publisher == nil || strings.TrimSpace(topic) == "" {
		return 0, 0, xerr.New(xerr.Unavailable, "activity template stats publisher is not configured")
	}
	items, err := s.runtime.ClaimActivityTemplateStatsOutbox(ctx, workerID, s.now().UTC().UnixMilli(), lockTTL, batchSize)
	if err != nil {
		return 0, 0, err
	}
	success := 0
	for _, item := range items {
		tag, tagErr := activitymq.EventTypeTag(item.EventType)
		if tagErr == nil {
			tagErr = publisher.SendSync(ctx, rocketmqx.Message{Topic: topic, Tag: tag, Keys: []string{item.OutboxID}, Body: item.Payload})
		}
		if tagErr != nil {
			_ = s.runtime.MarkActivityTemplateStatsOutboxFailed(ctx, item.OutboxID, tagErr.Error(), s.now().UTC().UnixMilli())
			continue
		}
		if err := s.runtime.MarkActivityTemplateStatsOutboxSent(ctx, item.OutboxID, s.now().UTC().UnixMilli()); err != nil {
			return len(items), success, err
		}
		success++
	}
	return len(items), success, nil
}

func (s *Service) requireRuntime() error {
	if s == nil || s.runtime == nil {
		return xerr.New(xerr.Unavailable, "activity template runtime repository is not configured")
	}
	return nil
}

func utcDay(ms int64) string { return time.UnixMilli(ms).UTC().Format("2006-01-02") }

func parseUTCDay(value string) (int64, error) {
	t, err := time.Parse("2006-01-02", strings.TrimSpace(value))
	if err != nil || t.Format("2006-01-02") != strings.TrimSpace(value) {
		return 0, xerr.New(xerr.InvalidArgument, "task_day must be UTC YYYY-MM-DD")
	}
	return t.UnixMilli(), nil
}