package app

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"time"

	"hyapp/pkg/appcode"
	"hyapp/pkg/logx"
	"hyapp/pkg/outboxpartition"
	"hyapp/pkg/rocketmqx"
	"hyapp/pkg/walletmq"
	"hyapp/services/wallet-service/internal/config"
	mysqlstorage "hyapp/services/wallet-service/internal/storage/mysql"
)

func (a *App) startWalletOutboxWorkers(ctx context.Context, workerName string, options config.OutboxWorkerConfig, partitions *outboxpartition.Partitions, producer *rocketmqx.Producer, topic string, includeEventTypes []string, excludeEventTypes []string) {
	if options.Concurrency <= 0 {
		options.Concurrency = 1
	}
	for workerIndex := 1; workerIndex <= options.Concurrency; workerIndex++ {
		workerID := fmt.Sprintf("%s-%s-%02d", workerName, a.nodeID, workerIndex)
		a.workers.Add(1)
		go func() {
			defer a.workers.Done()
			a.runWalletOutboxWorker(ctx, workerID, options, partitions, producer, topic, includeEventTypes, excludeEventTypes)
		}()
	}
}

func (a *App) runWalletOutboxWorker(ctx context.Context, workerID string, options config.OutboxWorkerConfig, partitions *outboxpartition.Partitions, producer *rocketmqx.Producer, topic string, includeEventTypes []string, excludeEventTypes []string) {
	ticker := time.NewTicker(options.PollInterval)
	defer ticker.Stop()
	for {
		processed, err := a.processWalletOutboxBatch(ctx, workerID, options, partitions, producer, topic, includeEventTypes, excludeEventTypes)
		if err != nil && !errors.Is(err, context.Canceled) {
			logx.Error(ctx, "wallet_outbox_publish_failed", err, slog.String("worker_id", workerID))
		}
		if processed >= options.BatchSize {
			continue
		}
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}

func (a *App) processWalletOutboxBatch(ctx context.Context, workerID string, options config.OutboxWorkerConfig, partitions *outboxpartition.Partitions, producer *rocketmqx.Producer, topic string, includeEventTypes []string, excludeEventTypes []string) (int, error) {
	if a.mysqlRepo == nil || producer == nil {
		return 0, nil
	}
	appCodes, discoveryErr := partitions.Resolve(ctx, a.mysqlRepo.ListWalletOutboxAppCodes)
	if discoveryErr != nil {
		if len(appCodes) == 0 {
			return 0, discoveryErr
		}
		// 已知 App 继续投递，刷新失败只影响新租户发现；缓存层会延后下一次查询，避免热循环打 MySQL。
		logx.Warn(ctx, "wallet_outbox_app_discovery_stale", slog.String("worker_id", workerID), slog.String("error", discoveryErr.Error()))
	}
	if len(appCodes) == 0 {
		return 0, nil
	}
	processed := 0
	for processed < options.BatchSize {
		// Sync publish is intentionally claimed one record at a time. Claiming an entire
		// batch with one short lock lets later rows expire while earlier SendSync calls
		// are still running, so another replica can publish the same tail concurrently.
		records, err := outboxpartition.ClaimFair(ctx, partitions.Order(appCodes), 1, func(claimCtx context.Context, appCode string, limit int) ([]mysqlstorage.WalletOutboxRecord, error) {
			return a.mysqlRepo.ClaimPendingWalletOutboxFiltered(appcode.WithContext(claimCtx, appCode), workerID, limit, time.Now().UTC().Add(walletOutboxSingleRecordLease(options.PublishTimeout)).UnixMilli(), mysqlstorage.WalletOutboxClaimFilter{
				IncludeEventTypes: includeEventTypes,
				ExcludeEventTypes: excludeEventTypes,
			})
		})
		if err != nil {
			return processed, err
		}
		if len(records) == 0 {
			return processed, nil
		}
		record := records[0]
		processed++
		if record.RetryCount >= options.MaxRetryCount {
			markCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
			err := a.mysqlRepo.MarkWalletOutboxDead(markCtx, record.EventID, deadWalletOutboxReason(record))
			cancel()
			if err != nil {
				return processed, err
			}
			continue
		}
		publishCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
		err = a.publishWalletOutboxRecord(publishCtx, producer, topic, record)
		cancel()
		if err != nil {
			nextRetryAtMS := time.Now().UTC().Add(walletOutboxBackoff(record.RetryCount+1, options)).UnixMilli()
			markCtx, markCancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
			markErr := a.mysqlRepo.MarkWalletOutboxRetryable(markCtx, record.EventID, err.Error(), nextRetryAtMS)
			markCancel()
			if markErr != nil {
				return processed, markErr
			}
			continue
		}
		markCtx, markCancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
		markErr := a.mysqlRepo.MarkWalletOutboxDelivered(markCtx, record.EventID)
		markCancel()
		if markErr != nil {
			return processed, markErr
		}
	}
	return processed, nil
}

func walletOutboxSingleRecordLease(publishTimeout time.Duration) time.Duration {
	if publishTimeout <= 0 {
		publishTimeout = 3 * time.Second
	}
	// 一条事实最多经历一次 publish deadline 和一次状态落库 deadline；再保留
	// 一个 deadline 作为调度余量。进程崩溃后的接管时间仍与单条消息耗时有界，
	// 不会随着 batch_size 放大成数十分钟。
	return 3 * publishTimeout
}

func (a *App) publishWalletOutboxRecord(ctx context.Context, producer *rocketmqx.Producer, topic string, record mysqlstorage.WalletOutboxRecord) error {
	message, err := walletOutboxRocketMessage(topic, a.walletEventTagMode, record)
	if err != nil {
		return err
	}
	return producer.SendSync(ctx, message)
}

// walletOutboxRocketMessage keeps the envelope event_type canonical in both phases.
// Phase A deliberately publishes the legacy Tag; Phase B uses that same event_type as
// the Broker routing Tag, avoiding any derived category that could drift from handlers.
func walletOutboxRocketMessage(topic string, tagMode config.WalletEventTagMode, record mysqlstorage.WalletOutboxRecord) (rocketmqx.Message, error) {
	eventTag, err := walletmq.EventTypeTag(record.EventType)
	if err != nil {
		return rocketmqx.Message{}, err
	}
	publishTag := walletmq.TagWalletOutboxEvent
	switch tagMode {
	case "", config.WalletEventTagModeLegacy:
		// 空值属于旧配置或测试构造，必须保持 legacy；只有显式 event_type 才能进入 Phase B。
	case config.WalletEventTagModeEventType:
		publishTag = eventTag
	default:
		return rocketmqx.Message{}, fmt.Errorf("unsupported wallet event tag mode %q", tagMode)
	}
	transactionID := strings.TrimSpace(record.TransactionID)
	if transactionID == "" && isHistoricalResourceOutboxRecord(record) {
		// 2026-07-11 以前的资源事实没有金融 transaction_id；event_id 已由
		// event_type+command_id+user_id+resource_id 稳定派生。发布时使用它作为
		// correlation id 与当前写入逻辑一致，避免为了回放改写已提交的 payload，
		// 同时严格限制为 RESOURCE 和已知资源事件，绝不替金融事实补造交易号。
		transactionID = strings.TrimSpace(record.EventID)
	}
	body, err := walletmq.EncodeWalletOutboxMessage(walletmq.WalletOutboxMessage{
		AppCode:        record.AppCode,
		EventID:        record.EventID,
		EventType:      eventTag,
		TransactionID:  transactionID,
		CommandID:      record.CommandID,
		UserID:         record.UserID,
		AssetType:      record.AssetType,
		AvailableDelta: record.AvailableDelta,
		FrozenDelta:    record.FrozenDelta,
		PayloadJSON:    record.PayloadJSON,
		OccurredAtMS:   record.CreatedAtMS,
	})
	if err != nil {
		return rocketmqx.Message{}, err
	}
	return rocketmqx.Message{
		Topic: topic,
		Tag:   publishTag,
		Keys:  []string{record.EventID, transactionID, record.CommandID},
		Body:  body,
	}, nil
}

func isHistoricalResourceOutboxRecord(record mysqlstorage.WalletOutboxRecord) bool {
	if strings.TrimSpace(record.EventID) == "" || !strings.EqualFold(strings.TrimSpace(record.AssetType), "RESOURCE") {
		return false
	}
	switch strings.TrimSpace(record.EventType) {
	case "GiftConfigChanged",
		"ResourceChanged",
		"ResourceGrantRevoked",
		"ResourceGranted",
		"ResourceGroupChanged",
		"ResourceGroupGranted",
		"UserResourceChanged",
		"VipEffectiveChanged",
		"VipTrialCardGranted":
		return true
	default:
		return false
	}
}

func deadWalletOutboxReason(record mysqlstorage.WalletOutboxRecord) string {
	last := strings.TrimSpace(record.LastError)
	if last == "" {
		return fmt.Sprintf("wallet outbox exceeded retry limit: retry_count=%d", record.RetryCount)
	}
	return last
}

func walletOutboxBackoff(retryCount int, options config.OutboxWorkerConfig) time.Duration {
	if retryCount <= 0 {
		return options.InitialBackoff
	}
	backoff := options.InitialBackoff
	for i := 1; i < retryCount; i++ {
		backoff *= 2
		if backoff >= options.MaxBackoff {
			return options.MaxBackoff
		}
	}
	return backoff
}

func (a *App) closeBackgroundWorkers() {
	if a.stopWorker != nil {
		a.stopWorker()
	}
	a.workers.Wait()
}