package app

import (
	"context"

	"hyapp/pkg/appcode"
	"hyapp/pkg/rocketmqx"
	servicemq "hyapp/pkg/servicekit/mq"
	"hyapp/pkg/walletmq"
	"hyapp/services/wallet-service/internal/config"
	walletservice "hyapp/services/wallet-service/internal/service/wallet"
)

func (a *App) startMQ() error {
	if err := servicemq.StartProducers([]*rocketmqx.Producer{a.outboxProducer, a.realtimeOutboxProducer}); err != nil {
		return err
	}
	if err := servicemq.StartConsumers([]*rocketmqx.Consumer{a.projectionConsumer}); err != nil {
		servicemq.ShutdownProducers([]*rocketmqx.Producer{a.outboxProducer, a.realtimeOutboxProducer})
		return err
	}
	return nil
}

func (a *App) shutdownMQ() {
	servicemq.ShutdownConsumers([]*rocketmqx.Consumer{a.projectionConsumer})
	servicemq.ShutdownProducers([]*rocketmqx.Producer{a.outboxProducer, a.realtimeOutboxProducer})
}

func newWalletProjectionConsumer(cfg config.Config, svc *walletservice.Service) (*rocketmqx.Consumer, error) {
	if svc == nil || !cfg.ProjectionWorker.Enabled {
		return nil, nil
	}
	consumer, err := rocketmqx.NewConsumer(rocketMQProjectionConsumerConfig(cfg.RocketMQ, cfg.ProjectionWorker))
	if err != nil {
		return nil, err
	}
	workerID := "wallet-projection-" + cfg.NodeID
	if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
		walletMessage, err := walletmq.DecodeWalletOutboxMessage(message.Body)
		if err != nil {
			return err
		}
		_, err = svc.ProcessWalletProjectionMessage(appcode.WithContext(ctx, walletMessage.AppCode), workerID, walletMessage, cfg.ProjectionWorker.LockTTL)
		return err
	}); err != nil {
		_ = consumer.Shutdown()
		return nil, err
	}
	return consumer, nil
}

func rocketMQProjectionConsumerConfig(cfg config.RocketMQConfig, projection config.ProjectionWorkerConfig) rocketmqx.ConsumerConfig {
	return rocketmqx.ConsumerConfig{
		EndpointConfig: rocketmqx.EndpointConfig{
			NameServers:      cfg.NameServers,
			NameServerDomain: cfg.NameServerDomain,
			AccessKey:        cfg.AccessKey,
			SecretKey:        cfg.SecretKey,
			SecurityToken:    cfg.SecurityToken,
			Namespace:        cfg.Namespace,
			VIPChannel:       cfg.VIPChannel,
		},
		GroupName:         projection.ConsumerGroup,
		MaxReconsumeTimes: projection.ConsumerMaxReconsumeTimes,
		ConsumePullBatch:  int32(projection.BatchSize),
	}
}

func rocketMQProducerConfig(cfg config.RocketMQConfig, groupName string) rocketmqx.ProducerConfig {
	return rocketmqx.ProducerConfig{
		EndpointConfig: rocketmqx.EndpointConfig{
			NameServers:      cfg.NameServers,
			NameServerDomain: cfg.NameServerDomain,
			AccessKey:        cfg.AccessKey,
			SecretKey:        cfg.SecretKey,
			SecurityToken:    cfg.SecurityToken,
			Namespace:        cfg.Namespace,
			VIPChannel:       cfg.VIPChannel,
		},
		GroupName:   groupName,
		SendTimeout: cfg.SendTimeout,
		Retry:       cfg.Retry,
	}
}

func shutdownProducers(producers ...*rocketmqx.Producer) {
	servicemq.ShutdownProducers(producers)
}