package app

import (
	"context"
	"errors"
	"fmt"
	"time"

	"hyapp/pkg/outboxpartition"
	"hyapp/pkg/outboxreplay"
	"hyapp/pkg/rocketmqx"
	servicemq "hyapp/pkg/servicekit/mq"
	"hyapp/services/game-service/internal/config"
	mysqlstorage "hyapp/services/game-service/internal/storage/mysql"
)

// ReplayGameOutboxOnce runs the ordinary game fact publisher for one App and a
// process-wide hard limit. It opens no gRPC listener and no downstream service
// connection, so an empty partition is a read-only no-op apart from MQ producer
// registration.
func ReplayGameOutboxOnce(ctx context.Context, cfg config.Config, scope outboxreplay.Scope) (outboxreplay.Result, error) {
	validatedScope, err := outboxreplay.NewScope(scope.AppCode, scope.Limit, 500)
	if err != nil {
		return outboxreplay.Result{}, err
	}
	if !cfg.RocketMQ.GameOutbox.Enabled {
		return outboxreplay.Result{}, errors.New("game RocketMQ outbox is not enabled")
	}
	repository, err := mysqlstorage.Open(ctx, cfg.MySQLDSN)
	if err != nil {
		return outboxreplay.Result{}, err
	}
	defer func() { _ = repository.Close() }()
	producer, err := rocketmqx.NewProducer(rocketMQProducerConfig(cfg.RocketMQ))
	if err != nil {
		return outboxreplay.Result{}, err
	}
	if err := servicemq.StartProducers([]*rocketmqx.Producer{producer}); err != nil {
		return outboxreplay.Result{}, err
	}
	defer servicemq.ShutdownProducers([]*rocketmqx.Producer{producer})

	replayConfig := cfg
	replayConfig.OutboxWorker.BatchSize = 1
	replayConfig.OutboxWorker.AppCodes = []string{validatedScope.AppCode}
	replayCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	replayApp := &App{
		repo:             repository,
		outboxProducer:   producer,
		outboxPartitions: outboxpartition.New(replayConfig.OutboxWorker.AppCodes, outboxpartition.DefaultDiscoveryRefreshInterval),
		outboxCtx:        replayCtx,
		outboxCancel:     cancel,
		cfg:              replayConfig,
	}
	workerID := fmt.Sprintf("ops-game-outbox-%s-%d", validatedScope.AppCode, time.Now().UTC().UnixMilli())
	return outboxreplay.Run(ctx, validatedScope, func(context.Context) (int, error) {
		// processOutboxBatch derives the same worker prefix as the live service;
		// attach the unique run ID to NodeID so lock ownership remains auditable.
		replayApp.cfg.NodeID = workerID
		return replayApp.processOutboxBatch()
	})
}