package broker

import (
	"encoding/json"
	"github.com/UpmeshLTD/urus-aio/infrastructure/logging"
	"github.com/UpmeshLTD/urus-aio/infrastructure/utils"
	"github.com/streadway/amqp"
	"math"
	"time"
)

type MessageBrokerBody struct {
	MessageId   string      `json:"messageId"`
	Message     interface{} `json:"message"`
	MessageType []string    `json:"messageType"`
}

type RabbitmqMessageBroker struct {
	Config MessageBrokerConfig
	Logger logging.Logger

	Conn    *amqp.Connection
	Channel *amqp.Channel

	started      bool
	stopped      bool
	topicName    string
	queueName    string
	consumerName string
	fn           ConsumeFn
}

func (broker *RabbitmqMessageBroker) Start() {
	if broker.started {
		return
	}
	broker.started = true

	err := broker.connect()
	if err != nil {
		panic(err)
	}

	broker.Logger.Info("Started")
}

func (broker *RabbitmqMessageBroker) Stop() {
	broker.stopped = true
	if broker.stopped {
		return
	}

	if broker.Channel != nil {
		if broker.consumerName != "" {
			broker.Channel.Cancel(broker.consumerName, false)
		}
		broker.Channel.Close()
	}
	if broker.Conn != nil {
		broker.Conn.Close()
	}

	broker.Channel = nil
	broker.Conn = nil

	broker.Logger.Info("Stopped")
}

func (broker *RabbitmqMessageBroker) Publish(topicName string, id string, body MessageBrokerBody) error {
	jsonBody, _ := json.Marshal(body)

	return broker.Channel.Publish(
		broker.Config.ExchangeName, // exchange
		topicName,                  // routing key
		false,                      // mandatory
		false,                      // immediate
		amqp.Publishing{
			ContentType: "application/vnd.masstransit+json",
			MessageId:   id,
			Body:        jsonBody,
		})
}

func (broker *RabbitmqMessageBroker) Subscribe(topicName, queueName string, fn ConsumeFn) {
	broker.consumerName = utils.NewId("consumer")
	broker.topicName = topicName
	broker.fn = fn

	queue, err := broker.Channel.QueueDeclare(
		queueName,                   // name
		true,                        // durable
		broker.Config.IsDevelopment, // delete when unused (ONLY in dev env)
		false,                       // exclusive
		false,                       // no-wait
		nil,                         // arguments
	)
	if err != nil {
		panic(err)
	}
	broker.queueName = queue.Name

	err = broker.Channel.QueueBind(
		queue.Name,                 // queue name
		topicName,                  // routing key
		broker.Config.ExchangeName, // exchange
		false,                      // no-wait
		nil,                        // arguments
	)
	if err != nil {
		panic(err)
	}

	delivery, err := broker.Channel.Consume(
		queue.Name,          // queue
		broker.consumerName, // consumer
		false,               // auto ack
		false,               // exclusive
		false,               // no local
		false,               // no wait
		nil,                 // args
	)
	if err != nil {
		panic(err)
	}

	go func() {
		for msg := range delivery {
			err := fn(msg.MessageId, msg.Body)

			if err == nil {
				_ = broker.Channel.Ack(msg.DeliveryTag, false)
			} else {
				broker.Logger.Error(err.Error(), "message_id", msg.MessageId)
				_ = broker.Channel.Nack(msg.DeliveryTag, false, false)
			}
		}
	}()
}

func (broker *RabbitmqMessageBroker) connect() error {
	conn, err := amqp.Dial(broker.Config.Uri)
	if err != nil {
		return err
	}
	broker.Conn = conn

	ch, err := conn.Channel()
	if err != nil {
		return err
	}

	err = ch.ExchangeDeclare(
		broker.Config.ExchangeName, // name
		"fanout",                   // type
		true,                       // durable
		false,                      // auto-deleted
		false,                      // internal
		false,                      // no-wait
		nil,                        // arguments
	)
	if err != nil {
		return err
	}
	broker.Channel = ch

	broker.reconnect()
	return nil
}

func (broker *RabbitmqMessageBroker) reconnect() {
	go func() {
		for {
			reason := <-broker.Channel.NotifyClose(make(chan *amqp.Error))
			broker.Logger.Error("got closed event from rabbitmq", "reason", reason)

			// if the close event was fired by ourself, ignore reconnect logic
			if broker.stopped {
				broker.Logger.Warn("programmatically closed")
				return
			}

			retry := 1
			for {
				t := math.Round(math.Log(float64(retry))) + 1
				delay := time.Duration(t) * time.Second
				broker.Logger.Warn("reconnecting...", "retry_count", retry, "delay_time", t)
				// wait 1s for connection reconnect
				time.Sleep(delay)
				retry++

				err := broker.connect()
				if err == nil {
					broker.Logger.Warn("connected OK", "retry_count", retry)

					// when we reconnected, if that is a subcribe app, we have to re-subcribe again
					if broker.topicName != "" {
						broker.Subscribe(broker.topicName, broker.queueName, broker.fn)
					}
					break
				} else {
					broker.Logger.Warn(err.Error(), "retry_count", retry)
				}
			}
		}
	}()
}