package app

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

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

// ReplayUserOutboxOnce processes a process-wide bounded App batch without
// starting user-service listeners or its unrelated consumers. Claiming one row
// per iteration keeps the existing 30-second row lease attached only to the
// message currently being published.
func ReplayUserOutboxOnce(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.UserOutbox.Enabled {
		return outboxreplay.Result{}, errors.New("user 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(userOutboxProducerConfig(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}
	workers := serviceapp.NewBackground(ctx)
	defer workers.StopAndWait()
	replayApp := &App{
		mysqlRepo:            repository,
		userOutboxProducer:   producer,
		userOutboxPartitions: outboxpartition.New(replayConfig.OutboxWorker.AppCodes, outboxpartition.DefaultDiscoveryRefreshInterval),
		cfg:                  replayConfig,
		workers:              workers,
	}
	workerID := fmt.Sprintf("ops-user-outbox-%s-%d", validatedScope.AppCode, time.Now().UTC().UnixMilli())
	return outboxreplay.Run(ctx, validatedScope, func(context.Context) (int, error) {
		return replayApp.processUserOutboxBatch(workerID)
	})
}