package app

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

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	healthgrpc "google.golang.org/grpc/health/grpc_health_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/walletmq"
	"hyapp/services/wallet-service/internal/client"
	"hyapp/services/wallet-service/internal/client/googleplay"
	"hyapp/services/wallet-service/internal/config"
	walletservice "hyapp/services/wallet-service/internal/service/wallet"
	mysqlstorage "hyapp/services/wallet-service/internal/storage/mysql"
	grpcserver "hyapp/services/wallet-service/internal/transport/grpc"
)

// App 装配 wallet-service gRPC 入口。
type App struct {
	server                    *grpc.Server
	listener                  net.Listener
	health                    *grpchealth.ServingChecker
	healthHTTP                *healthhttp.Server
	mysqlRepo                 *mysqlstorage.Repository
	walletSvc                 *walletservice.Service
	activityConn              *grpc.ClientConn
	outboxProducer            *rocketmqx.Producer
	realtimeOutboxProducer    *rocketmqx.Producer
	projectionConsumer        *rocketmqx.Consumer
	outboxWorkerCfg           config.OutboxWorkerConfig
	realtimeOutboxWorkerCfg   config.OutboxWorkerConfig
	projectionWorkerCfg       config.ProjectionWorkerConfig
	walletOutboxTopic         string
	realtimeWalletOutboxTopic string
	nodeID                    string
	redPacketExpiryWorkerCfg  config.RedPacketExpiryWorkerConfig
	stopWorker                context.CancelFunc
	workers                   sync.WaitGroup
	closeOnce                 sync.Once
}

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

	// wallet-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
	}
	activityConn, err := grpc.Dial(cfg.ActivityServiceAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		_ = listener.Close()
		_ = repository.Close()
		return nil, err
	}
	var outboxProducer *rocketmqx.Producer
	if cfg.RocketMQ.WalletOutbox.Enabled {
		outboxProducer, err = rocketmqx.NewProducer(rocketMQProducerConfig(cfg.RocketMQ, cfg.RocketMQ.WalletOutbox.ProducerGroup))
		if err != nil {
			_ = activityConn.Close()
			_ = listener.Close()
			_ = repository.Close()
			return nil, err
		}
	}
	var realtimeOutboxProducer *rocketmqx.Producer
	if cfg.RocketMQ.RealtimeOutbox.Enabled {
		realtimeOutboxProducer, err = rocketmqx.NewProducer(rocketMQProducerConfig(cfg.RocketMQ, cfg.RocketMQ.RealtimeOutbox.ProducerGroup))
		if err != nil {
			shutdownProducers(outboxProducer)
			_ = activityConn.Close()
			_ = listener.Close()
			_ = repository.Close()
			return nil, err
		}
	}

	server := grpc.NewServer(grpc.UnaryInterceptor(logx.UnaryServerInterceptor("wallet-service")))
	svc := walletservice.New(repository, client.NewActivityAchievementClient(activityConn))
	if cfg.GooglePlay.Enabled {
		googleClient, err := googleplay.New(cfg.GooglePlay)
		if err != nil {
			shutdownProducers(outboxProducer, realtimeOutboxProducer)
			_ = activityConn.Close()
			_ = listener.Close()
			_ = repository.Close()
			return nil, err
		}
		svc.SetGooglePlayClient(googleClient)
	}
	walletv1.RegisterWalletServiceServer(server, grpcserver.NewServer(svc))
	walletv1.RegisterWalletCronServiceServer(server, grpcserver.NewCronServer(svc))
	health := grpchealth.NewServingChecker("wallet-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 {
		shutdownProducers(outboxProducer, realtimeOutboxProducer)
		_ = activityConn.Close()
		_ = listener.Close()
		_ = repository.Close()
		return nil, err
	}
	projectionConsumer, err := newWalletProjectionConsumer(cfg, svc)
	if err != nil {
		shutdownProducers(outboxProducer, realtimeOutboxProducer)
		_ = activityConn.Close()
		_ = listener.Close()
		_ = repository.Close()
		return nil, err
	}

	return &App{
		server:                    server,
		listener:                  listener,
		health:                    health,
		healthHTTP:                healthHTTP,
		mysqlRepo:                 repository,
		walletSvc:                 svc,
		activityConn:              activityConn,
		outboxProducer:            outboxProducer,
		realtimeOutboxProducer:    realtimeOutboxProducer,
		projectionConsumer:        projectionConsumer,
		outboxWorkerCfg:           cfg.OutboxWorker,
		realtimeOutboxWorkerCfg:   cfg.RealtimeOutboxWorker,
		projectionWorkerCfg:       cfg.ProjectionWorker,
		walletOutboxTopic:         cfg.RocketMQ.WalletOutbox.Topic,
		realtimeWalletOutboxTopic: cfg.RocketMQ.RealtimeOutbox.Topic,
		nodeID:                    cfg.NodeID,
		redPacketExpiryWorkerCfg:  cfg.RedPacketExpiryWorker,
	}, nil
}

// Run 启动 gRPC 服务。
func (a *App) Run() error {
	a.runHealthHTTP()
	if err := a.startMQ(); err != nil {
		a.health.MarkStopped()
		return err
	}
	a.runBackgroundWorkers()
	a.health.MarkServing()
	err := a.server.Serve(a.listener)
	a.health.MarkStopped()
	a.closeBackgroundWorkers()
	a.shutdownMQ()
	if errors.Is(err, grpc.ErrServerStopped) {
		return nil
	}

	return err
}

// Close 关闭 gRPC 服务。
func (a *App) Close() {
	a.closeOnce.Do(func() {
		a.health.MarkDraining()
		a.closeBackgroundWorkers()
		grpcshutdown.GracefulStop(a.server, 15*time.Second)
		a.closeHealthHTTP()
		if a.activityConn != nil {
			_ = a.activityConn.Close()
		}
		a.shutdownMQ()
		if a.mysqlRepo != nil {
			// MySQL 连接池最后关闭，避免正在 drain 的扣费请求丢失提交能力。
			_ = a.mysqlRepo.Close()
		}
	})
}

func (a *App) runBackgroundWorkers() {
	if a.walletSvc == nil {
		return
	}
	ctx, cancel := context.WithCancel(context.Background())
	a.stopWorker = cancel
	if a.outboxWorkerCfg.Enabled && a.outboxProducer != nil {
		excludedRealtimeTypes := []string(nil)
		if a.realtimeOutboxWorkerCfg.Enabled && a.realtimeOutboxProducer != nil {
			// 普通账务 worker 在实时通道可用时主动跳过红包 UI 事件，避免两个 worker 抢同一行，
			// 也避免红包事实继续排在 WalletBalanceChanged 和礼物流水之后。
			excludedRealtimeTypes = a.realtimeOutboxWorkerCfg.EventTypes
		}
		a.startWalletOutboxWorkers(ctx, "wallet-outbox", a.outboxWorkerCfg, a.outboxProducer, a.walletOutboxTopic, nil, excludedRealtimeTypes)
	}
	if a.realtimeOutboxWorkerCfg.Enabled && a.realtimeOutboxProducer != nil {
		a.startWalletOutboxWorkers(ctx, "wallet-realtime-outbox", a.realtimeOutboxWorkerCfg, a.realtimeOutboxProducer, a.realtimeWalletOutboxTopic, a.realtimeOutboxWorkerCfg.EventTypes, nil)
	}
	if a.redPacketExpiryWorkerCfg.Enabled {
		redPacketWorkerID := "red-packet-expiry-" + a.nodeID
		a.workers.Add(1)
		go func() {
			defer a.workers.Done()
			ticker := time.NewTicker(a.redPacketExpiryWorkerCfg.PollInterval)
			defer ticker.Stop()
			for {
				result, err := a.walletSvc.ExpireRedPackets(ctx, a.redPacketExpiryWorkerCfg.AppCode, a.redPacketExpiryWorkerCfg.BatchSize)
				if err != nil && !errors.Is(err, context.Canceled) {
					logx.Error(ctx, "red_packet_expiry_failed", err, slog.String("worker_id", redPacketWorkerID))
				}
				if result.ExpiredCount > 0 {
					logx.Info(ctx, "red_packet_expired",
						slog.String("worker_id", redPacketWorkerID),
						slog.Int64("expired_count", int64(result.ExpiredCount)),
						slog.Int64("refunded_amount", result.RefundedAmount),
					)
				}
				if result.ExpiredCount >= a.redPacketExpiryWorkerCfg.BatchSize {
					continue
				}
				select {
				case <-ctx.Done():
					return
				case <-ticker.C:
				}
			}
		}()
	}
}

func (a *App) startWalletOutboxWorkers(ctx context.Context, workerName string, options config.OutboxWorkerConfig, 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, producer, topic, includeEventTypes, excludeEventTypes)
		}()
	}
}

func (a *App) runWalletOutboxWorker(ctx context.Context, workerID string, options config.OutboxWorkerConfig, 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, 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, producer *rocketmqx.Producer, topic string, includeEventTypes []string, excludeEventTypes []string) (int, error) {
	if a.mysqlRepo == nil || producer == nil {
		return 0, nil
	}
	records, err := a.mysqlRepo.ClaimPendingWalletOutboxFiltered(ctx, workerID, options.BatchSize, time.Now().UTC().Add(options.PublishTimeout).UnixMilli(), mysqlstorage.WalletOutboxClaimFilter{
		IncludeEventTypes: includeEventTypes,
		ExcludeEventTypes: excludeEventTypes,
	})
	if err != nil {
		return 0, err
	}
	for _, record := range records {
		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 0, 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 0, 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 0, markErr
		}
	}
	return len(records), nil
}

func (a *App) publishWalletOutboxRecord(ctx context.Context, producer *rocketmqx.Producer, topic string, record mysqlstorage.WalletOutboxRecord) error {
	body, err := walletmq.EncodeWalletOutboxMessage(walletmq.WalletOutboxMessage{
		AppCode:        record.AppCode,
		EventID:        record.EventID,
		EventType:      record.EventType,
		TransactionID:  record.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 err
	}
	return producer.SendSync(ctx, rocketmqx.Message{
		Topic: topic,
		Tag:   walletmq.TagWalletOutboxEvent,
		Keys:  []string{record.EventID, record.TransactionID, record.CommandID},
		Body:  body,
	})
}

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()
}

func (a *App) startMQ() error {
	if a.outboxProducer != nil {
		if err := a.outboxProducer.Start(); err != nil {
			return err
		}
	}
	if a.realtimeOutboxProducer != nil {
		if err := a.realtimeOutboxProducer.Start(); err != nil {
			if a.outboxProducer != nil {
				_ = a.outboxProducer.Shutdown()
			}
			return err
		}
	}
	if a.projectionConsumer != nil {
		if err := a.projectionConsumer.Start(); err != nil {
			shutdownProducers(a.outboxProducer, a.realtimeOutboxProducer)
			return err
		}
	}
	return nil
}

func (a *App) shutdownMQ() {
	if a.projectionConsumer != nil {
		if err := a.projectionConsumer.Shutdown(); err != nil {
			logx.Warn(context.Background(), "rocketmq_projection_consumer_shutdown_failed", slog.String("error", err.Error()))
		}
	}
	if a.outboxProducer != nil {
		if err := a.outboxProducer.Shutdown(); err != nil {
			logx.Warn(context.Background(), "rocketmq_producer_shutdown_failed", slog.String("error", err.Error()))
		}
	}
	if a.realtimeOutboxProducer != nil {
		if err := a.realtimeOutboxProducer.Shutdown(); err != nil {
			logx.Warn(context.Background(), "rocketmq_realtime_producer_shutdown_failed", slog.String("error", err.Error()))
		}
	}
}

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) {
	for _, producer := range producers {
		if producer != nil {
			_ = producer.Shutdown()
		}
	}
}

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)
}