// Code generated by 'freedom new-project infra-example'
package main

import (
	"time"

	"github.com/8treenet/freedom"
	_ "github.com/8treenet/freedom/example/infra-example/adapter/controller"
	_ "github.com/8treenet/freedom/example/infra-example/adapter/repository"
	"github.com/8treenet/freedom/example/infra-example/config"
	"github.com/8treenet/freedom/example/infra-example/infra/domainevent"
	"github.com/8treenet/freedom/infra/kafka"
	"github.com/8treenet/freedom/infra/requests"
	"github.com/8treenet/freedom/middleware"
	"github.com/Shopify/sarama"
	"github.com/prometheus/client_golang/prometheus"
	"gorm.io/driver/mysql"
	"gorm.io/gorm"
)

// mac: zookeeper-server-start /usr/local/etc/kafka/zookeeper.properties & kafka-server-start /usr/local/etc/kafka/server.properties
func main() {
	app := freedom.NewApplication()
	installDatabase(app)
	installMiddleware(app)
	addr := config.Get().App.Other["listen_addr"].(string)
	addrRunner := app.NewRunner(addr)
	//app.InstallParty("/example")
	liveness(app)

	go func() {
		//领域事件重试
		time.Sleep(5 * time.Second)
		timer := time.NewTimer(5 * time.Minute)
		for range timer.C {
			domainevent.GetEventManager().Retry()
			timer.Reset(5 * time.Minute)
		}
	}()

	kafkaConf := sarama.NewConfig()
	kafkaConf.Version = sarama.V0_11_0_2
	kafka.GetConsumer().Start([]string{":9092"}, "freedom1", kafkaConf, "http://127.0.0.1:8000", false)
	kafka.GetProducer().Start([]string{":9092"}, kafkaConf)
	app.Run(addrRunner, config.Get().App)
}

func installMiddleware(app freedom.Application) {
	//Recover中间件
	app.InstallMiddleware(middleware.NewRecover())
	//Trace链路中间件
	app.InstallMiddleware(middleware.NewTrace("x-request-id"))
	//日志中间件，每个请求一个logger
	app.InstallMiddleware(middleware.NewRequestLogger("x-request-id"))
	//logRow中间件，每一行日志都会触发回调。如果返回true，将停止中间件遍历回调。
	app.Logger().Handle(middleware.DefaultLogRowHandle)

	//HttpClient 普罗米修斯中间件，监控ClientAPI的请求。
	middle := middleware.NewClientPrometheus(config.Get().App.Other["service_name"].(string), freedom.Prometheus())
	requests.InstallMiddleware(middle)

	//总线中间件，处理上下游透传的Header
	app.InstallBusMiddleware(middleware.NewBusFilter())

	//安装消息监控中间件
	eventMiddle := NewMsgPrometheus(config.Get().App.Other["service_name"].(string))
	kafka.InstallMiddleware(eventMiddle)

	//注册服务定位器的回调函数, freedom.ServiceLocator().Call 之前触发
	freedom.ServiceLocator().InstallBeginCallBack(func(worker freedom.Worker) {
		traceID, _ := middleware.GenerateTraceID()
		logger := middleware.NewLogger("x-request-id", traceID)
		worker.Bus().Set("x-request-id", traceID)
		worker.SetLogger(logger)
	})
}

func installDatabase(app freedom.Application) {
	app.InstallDB(func() interface{} {
		conf := config.Get().DB
		db, err := gorm.Open(mysql.Open(conf.Addr), &gorm.Config{})
		if err != nil {
			freedom.Logger().Fatal(err.Error())
		}

		sqlDB, err := db.DB()
		if err != nil {
			freedom.Logger().Fatal(err)
		}
		if err = sqlDB.Ping(); err != nil {
			freedom.Logger().Fatal(err)
		}

		sqlDB.SetMaxIdleConns(conf.MaxIdleConns)
		sqlDB.SetMaxOpenConns(conf.MaxOpenConns)
		sqlDB.SetConnMaxLifetime(time.Duration(conf.ConnMaxLifeTime) * time.Second)
		return db
	})
}

func liveness(app freedom.Application) {
	app.Iris().Get("/ping", func(ctx freedom.Context) {
		ctx.WriteString("pong")
	})
}

// NewMsgPrometheus Kafka消息中间件
func NewMsgPrometheus(serviceName string) kafka.ProducerHandler {
	msgPushReqs := prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name:        "kafka_producer_total",
			Help:        "监控Kafka消息发送总数",
			ConstLabels: prometheus.Labels{"service": serviceName},
		},
		[]string{"topic", "error"},
	)
	msgPushLatency := prometheus.NewHistogramVec(prometheus.HistogramOpts{
		Name:        "kafka_producer_duration_seconds",
		Help:        "监控Kafka消息发送耗时",
		ConstLabels: prometheus.Labels{"service": serviceName},
	},
		[]string{"topic", "error"},
	)

	freedom.Prometheus().RegisterCounter(msgPushReqs)
	freedom.Prometheus().RegisterHistogram(msgPushLatency)

	return func(msg *kafka.Msg) {
		//返回一个消息中间件
		if msg.IsStopped() {
			//如果停止
			return
		}
		start := time.Now()
		//下一步
		msg.Next()

		//如果发送错误
		if msg.GetExecution() != nil {
			msgPushReqs.WithLabelValues(msg.Topic, msg.GetExecution().Error()).Inc()
			msgPushLatency.WithLabelValues(msg.Topic, msg.GetExecution().Error()).
				Observe(float64(time.Since(start).Nanoseconds()) / 1000000000)
			return
		}
		msgPushReqs.WithLabelValues(msg.Topic, "").Inc()
		msgPushLatency.WithLabelValues(msg.Topic, "").
			Observe(float64(time.Since(start).Nanoseconds()) / 1000000000)
	}
}
