package app

import (
	"context"
	"errors"
	"log/slog"
	"net"
	"sync"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	healthgrpc "google.golang.org/grpc/health/grpc_health_v1"
	activityv1 "hyapp.local/api/proto/activity/v1"
	walletv1 "hyapp.local/api/proto/wallet/v1"
	"hyapp/pkg/appcode"
	"hyapp/pkg/grpchealth"
	"hyapp/pkg/grpcshutdown"
	"hyapp/pkg/healthhttp"
	"hyapp/pkg/logx"
	"hyapp/pkg/rocketmqx"
	"hyapp/pkg/roommq"
	"hyapp/pkg/tencentim"
	"hyapp/services/activity-service/internal/client"
	"hyapp/services/activity-service/internal/config"
	achievementservice "hyapp/services/activity-service/internal/service/achievement"
	activityservice "hyapp/services/activity-service/internal/service/activity"
	broadcastservice "hyapp/services/activity-service/internal/service/broadcast"
	growthservice "hyapp/services/activity-service/internal/service/growth"
	luckygiftservice "hyapp/services/activity-service/internal/service/luckygift"
	messageservice "hyapp/services/activity-service/internal/service/message"
	registrationrewardservice "hyapp/services/activity-service/internal/service/registrationreward"
	sevendaycheckinservice "hyapp/services/activity-service/internal/service/sevendaycheckin"
	taskservice "hyapp/services/activity-service/internal/service/task"
	mysqlstorage "hyapp/services/activity-service/internal/storage/mysql"
	grpcserver "hyapp/services/activity-service/internal/transport/grpc"
)

// App 装配 activity-service gRPC 入口和底座依赖。
type App struct {
	server                 *grpc.Server
	listener               net.Listener
	health                 *grpchealth.ServingChecker
	healthHTTP             *healthhttp.Server
	mysqlRepo              *mysqlstorage.Repository
	broadcast              *broadcastservice.Service
	broadcastWorkerEnabled bool
	mqConsumers            []*rocketmqx.Consumer
	userConn               *grpc.ClientConn
	walletConn             *grpc.ClientConn
	workerCtx              context.Context
	workerStop             context.CancelFunc
	workerWG               sync.WaitGroup
	closeOnce              sync.Once
}

// New 初始化 activity-service 应用。
func New(cfg config.Config) (*App, error) {
	startupCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	// activity-service 的活动状态、消费幂等和 outbox 都必须使用 MySQL。
	repository, err := mysqlstorage.Open(startupCtx, cfg.MySQLDSN)
	if err != nil {
		return nil, err
	}

	listener, err := net.Listen("tcp", cfg.GRPCAddr)
	if err != nil {
		_ = repository.Close()
		return nil, err
	}
	userConn, err := grpc.Dial(cfg.UserServiceAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		_ = listener.Close()
		_ = repository.Close()
		return nil, err
	}
	walletConn, err := grpc.Dial(cfg.WalletServiceAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		_ = userConn.Close()
		_ = listener.Close()
		_ = repository.Close()
		return nil, err
	}

	server := grpc.NewServer(grpc.UnaryInterceptor(logx.UnaryServerInterceptor("activity-service")))
	svc := activityservice.New(activityservice.Config{NodeID: cfg.NodeID}, repository)
	messageSvc := messageservice.New(messageservice.Config{NodeID: cfg.NodeID}, repository, messageservice.WithTargetSource(client.NewGRPCUserTargetSource(userConn)))
	taskSvc := taskservice.New(repository, walletv1.NewWalletServiceClient(walletConn))
	registrationRewardSvc := registrationrewardservice.New(repository, walletv1.NewWalletServiceClient(walletConn))
	sevenDayCheckInSvc := sevendaycheckinservice.New(repository, walletv1.NewWalletServiceClient(walletConn))
	growthSvc := growthservice.New(repository, walletv1.NewWalletServiceClient(walletConn))
	achievementSvc := achievementservice.New(repository, walletv1.NewWalletServiceClient(walletConn))
	luckyGiftSvc := luckygiftservice.New(repository)
	var broadcastPublisher broadcastservice.Publisher
	if cfg.TencentIM.Enabled {
		tencentClient, err := tencentim.NewRESTClient(cfg.TencentIM.RESTConfig())
		if err != nil {
			_ = walletConn.Close()
			_ = userConn.Close()
			_ = listener.Close()
			_ = repository.Close()
			return nil, err
		}
		broadcastPublisher = tencentClient
	}
	if cfg.Broadcast.Enabled && broadcastPublisher == nil {
		_ = walletConn.Close()
		_ = userConn.Close()
		_ = listener.Close()
		_ = repository.Close()
		return nil, errors.New("broadcast worker requires tencent_im.enabled")
	}
	broadcastSvc := broadcastservice.New(broadcastservice.Config{
		NodeID:                cfg.NodeID,
		SuperGiftMinValue:     cfg.Broadcast.SuperGiftMinValue,
		WorkerBatchSize:       cfg.Broadcast.WorkerBatchSize,
		WorkerLockTTL:         cfg.Broadcast.WorkerLockTTL,
		WorkerMaxRetry:        cfg.Broadcast.WorkerMaxRetry,
		WorkerPollInterval:    cfg.Broadcast.WorkerPollInterval,
		EnsureGroupsOnStartup: cfg.Broadcast.EnsureGroupsOnStartup,
	}, repository, broadcastPublisher, client.NewGRPCRegionSource(userConn))
	activityv1.RegisterActivityServiceServer(server, grpcserver.NewServer(svc))
	activityv1.RegisterMessageInboxServiceServer(server, grpcserver.NewMessageServer(messageSvc))
	activityv1.RegisterActivityCronServiceServer(server, grpcserver.NewCronServer(messageSvc, growthSvc, achievementSvc))
	activityv1.RegisterTaskServiceServer(server, grpcserver.NewTaskServer(taskSvc))
	activityv1.RegisterAdminTaskServiceServer(server, grpcserver.NewAdminTaskServer(taskSvc))
	activityv1.RegisterGrowthLevelServiceServer(server, grpcserver.NewGrowthLevelServer(growthSvc))
	activityv1.RegisterAdminGrowthLevelServiceServer(server, grpcserver.NewAdminGrowthLevelServer(growthSvc))
	activityv1.RegisterAchievementServiceServer(server, grpcserver.NewAchievementServer(achievementSvc))
	activityv1.RegisterAdminAchievementServiceServer(server, grpcserver.NewAdminAchievementServer(achievementSvc))
	activityv1.RegisterLuckyGiftServiceServer(server, grpcserver.NewLuckyGiftServer(luckyGiftSvc))
	activityv1.RegisterAdminLuckyGiftServiceServer(server, grpcserver.NewAdminLuckyGiftServer(luckyGiftSvc))
	activityv1.RegisterRegistrationRewardServiceServer(server, grpcserver.NewRegistrationRewardServer(registrationRewardSvc))
	activityv1.RegisterAdminRegistrationRewardServiceServer(server, grpcserver.NewAdminRegistrationRewardServer(registrationRewardSvc))
	activityv1.RegisterSevenDayCheckInServiceServer(server, grpcserver.NewSevenDayCheckInServer(sevenDayCheckInSvc))
	activityv1.RegisterAdminSevenDayCheckInServiceServer(server, grpcserver.NewAdminSevenDayCheckInServer(sevenDayCheckInSvc))
	broadcastServer := grpcserver.NewBroadcastServer(broadcastSvc, growthSvc)
	activityv1.RegisterBroadcastServiceServer(server, broadcastServer)
	activityv1.RegisterRoomEventConsumerServiceServer(server, broadcastServer)
	mqConsumers := make([]*rocketmqx.Consumer, 0, 1)
	if cfg.RocketMQ.RoomOutbox.Enabled {
		consumer, err := rocketmqx.NewConsumer(rocketMQConsumerConfig(cfg.RocketMQ))
		if err != nil {
			_ = walletConn.Close()
			_ = userConn.Close()
			_ = listener.Close()
			_ = repository.Close()
			return nil, err
		}
		if err := consumer.Subscribe(cfg.RocketMQ.RoomOutbox.Topic, roommq.TagRoomOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
			envelope, _, err := roommq.DecodeRoomOutboxMessage(message.Body)
			if err != nil {
				return err
			}
			eventCtx := appcode.WithContext(ctx, envelope.GetAppCode())
			if _, err := broadcastSvc.HandleRoomEvent(eventCtx, envelope); err != nil {
				return err
			}
			_, err = growthSvc.HandleRoomEvent(eventCtx, envelope)
			return err
		}); err != nil {
			_ = walletConn.Close()
			_ = userConn.Close()
			_ = listener.Close()
			_ = repository.Close()
			return nil, err
		}
		mqConsumers = append(mqConsumers, consumer)
	}
	health := grpchealth.NewServingChecker("activity-service", grpchealth.Dependency{
		Name:  "mysql",
		Check: repository.Ping,
	})
	healthgrpc.RegisterHealthServer(server, grpchealth.NewServer(health))
	healthHTTP, err := healthhttp.New(cfg.HealthHTTPAddr, cfg.NodeID, health)
	if err != nil {
		_ = walletConn.Close()
		_ = userConn.Close()
		_ = listener.Close()
		_ = repository.Close()
		return nil, err
	}
	workerCtx, workerStop := context.WithCancel(context.Background())

	return &App{
		server:                 server,
		listener:               listener,
		health:                 health,
		healthHTTP:             healthHTTP,
		mysqlRepo:              repository,
		broadcast:              broadcastSvc,
		broadcastWorkerEnabled: cfg.Broadcast.Enabled,
		mqConsumers:            mqConsumers,
		userConn:               userConn,
		walletConn:             walletConn,
		workerCtx:              workerCtx,
		workerStop:             workerStop,
	}, nil
}

// Run 启动 gRPC 服务。
func (a *App) Run() error {
	a.runHealthHTTP()
	a.health.MarkServing()
	if err := a.startMQ(); err != nil {
		a.health.MarkStopped()
		return err
	}
	if a.broadcastWorkerEnabled && a.broadcast != nil && a.workerCtx != nil {
		a.workerWG.Go(func() {
			a.broadcast.RunWorker(a.workerCtx, broadcastservice.WorkerOptions{})
		})
	}
	err := a.server.Serve(a.listener)
	if a.workerStop != nil {
		a.workerStop()
	}
	a.workerWG.Wait()
	a.shutdownMQ()
	a.health.MarkStopped()
	if errors.Is(err, grpc.ErrServerStopped) {
		return nil
	}

	return err
}

// Close 关闭 gRPC 服务。
func (a *App) Close() {
	a.closeOnce.Do(func() {
		a.health.MarkDraining()
		if a.workerStop != nil {
			a.workerStop()
		}
		a.workerWG.Wait()
		a.shutdownMQ()
		grpcshutdown.GracefulStop(a.server, 15*time.Second)
		a.closeHealthHTTP()
		if a.userConn != nil {
			_ = a.userConn.Close()
		}
		if a.walletConn != nil {
			_ = a.walletConn.Close()
		}
		if a.mysqlRepo != nil {
			// MySQL 连接池最后关闭，保证 drain 中的查询或消费提交能完成。
			_ = a.mysqlRepo.Close()
		}
	})
}

func (a *App) runHealthHTTP() {
	if a.healthHTTP == nil {
		return
	}
	go func() {
		if err := a.healthHTTP.Run(); err != nil {
			logx.Error(context.Background(), "health_http_run_failed", err)
		}
	}()
}

func (a *App) closeHealthHTTP() {
	if a.healthHTTP == nil {
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	_ = a.healthHTTP.Close(ctx)
}

func (a *App) startMQ() error {
	for _, consumer := range a.mqConsumers {
		if err := consumer.Start(); err != nil {
			a.shutdownMQ()
			return err
		}
	}
	return nil
}

func (a *App) shutdownMQ() {
	for _, consumer := range a.mqConsumers {
		if err := consumer.Shutdown(); err != nil {
			logx.Warn(context.Background(), "rocketmq_consumer_shutdown_failed", slog.String("error", err.Error()))
		}
	}
}

func rocketMQConsumerConfig(cfg config.RocketMQConfig) 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:          cfg.RoomOutbox.ConsumerGroup,
		MaxReconsumeTimes:  cfg.RoomOutbox.ConsumerMaxReconsumeTimes,
		ConsumeRetryDelay:  time.Second,
		ConsumePullBatch:   32,
		ConsumePullTimeout: 15 * time.Minute,
	}
}