/*
 * Copyright 2018-present Open Networking Foundation

 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at

 * http://www.apache.org/licenses/LICENSE-2.0

 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package kafka

import (
	"errors"
	"fmt"
	"github.com/Shopify/sarama"
	scc "github.com/bsm/sarama-cluster"
	"github.com/golang/protobuf/proto"
	"github.com/opencord/voltha-go/common/log"
	ca "github.com/opencord/voltha-go/protos/core_adapter"
	"sync"
	"time"
)

// consumerChannels represents a consumer listening on a kafka topic.  Once it receives a message on that
// topic it broadcasts the message to all the listening channels
type consumerChannels struct {
	consumer sarama.PartitionConsumer
	//consumer *sc.Consumer
	channels []chan *ca.InterContainerMessage
}

// SaramaClient represents the messaging proxy
type SaramaClient struct {
	broker                        *sarama.Broker
	client                        sarama.Client
	KafkaHost                     string
	KafkaPort                     int
	producer                      sarama.AsyncProducer
	consumer                      sarama.Consumer
	groupConsumer                 *scc.Consumer
	doneCh                        chan int
	topicToConsumerChannelMap     map[string]*consumerChannels
	lockTopicToConsumerChannelMap sync.RWMutex
}

type SaramaClientOption func(*SaramaClient)

func Host(host string) SaramaClientOption {
	return func(args *SaramaClient) {
		args.KafkaHost = host
	}
}

func Port(port int) SaramaClientOption {
	return func(args *SaramaClient) {
		args.KafkaPort = port
	}
}

func NewSaramaClient(opts ...SaramaClientOption) *SaramaClient {
	client := &SaramaClient{
		KafkaHost: DefaultKafkaHost,
		KafkaPort: DefaultKafkaPort,
	}

	for _, option := range opts {
		option(client)
	}

	client.lockTopicToConsumerChannelMap = sync.RWMutex{}

	return client
}

func (sc *SaramaClient) Start(retries int) error {
	log.Info("Starting-Proxy")

	// Create the Done channel
	sc.doneCh = make(chan int, 1)

	// Create the Publisher
	if err := sc.createPublisher(retries); err != nil {
		log.Errorw("Cannot-create-kafka-publisher", log.Fields{"error": err})
		return err
	}

	// Create the master consumer
	if err := sc.createConsumer(retries); err != nil {
		log.Errorw("Cannot-create-kafka-consumer", log.Fields{"error": err})
		return err
	}

	// Create the topic to consumer/channel map
	sc.topicToConsumerChannelMap = make(map[string]*consumerChannels)

	return nil
}

func (sc *SaramaClient) Stop() {
	log.Info("stopping-sarama-client")

	//Send a message over the done channel to close all long running routines
	sc.doneCh <- 1

	// Clear the consumer map
	//sc.clearConsumerChannelMap()

	if sc.producer != nil {
		if err := sc.producer.Close(); err != nil {
			panic(err)
		}
	}
	if sc.consumer != nil {
		if err := sc.consumer.Close(); err != nil {
			panic(err)
		}
	}
}

func (sc *SaramaClient) addTopicToConsumerChannelMap(id string, arg *consumerChannels) {
	sc.lockTopicToConsumerChannelMap.Lock()
	defer sc.lockTopicToConsumerChannelMap.Unlock()
	if _, exist := sc.topicToConsumerChannelMap[id]; !exist {
		sc.topicToConsumerChannelMap[id] = arg
	}
}

func (sc *SaramaClient) deleteFromTopicToConsumerChannelMap(id string) {
	sc.lockTopicToConsumerChannelMap.Lock()
	defer sc.lockTopicToConsumerChannelMap.Unlock()
	if _, exist := sc.topicToConsumerChannelMap[id]; exist {
		delete(sc.topicToConsumerChannelMap, id)
	}
}

func (sc *SaramaClient) getConsumerChannel(topic Topic) *consumerChannels {
	sc.lockTopicToConsumerChannelMap.Lock()
	defer sc.lockTopicToConsumerChannelMap.Unlock()

	if consumerCh, exist := sc.topicToConsumerChannelMap[topic.Name]; exist {
		return consumerCh
	}
	return nil
}

func (sc *SaramaClient) addChannelToConsumerChannelMap(topic Topic, ch chan *ca.InterContainerMessage) {
	sc.lockTopicToConsumerChannelMap.Lock()
	defer sc.lockTopicToConsumerChannelMap.Unlock()
	if consumerCh, exist := sc.topicToConsumerChannelMap[topic.Name]; exist {
		consumerCh.channels = append(consumerCh.channels, ch)
		return
	}
	log.Warnw("consumer-channel-not-exist", log.Fields{"topic": topic.Name})
}

func (sc *SaramaClient) removeChannelFromConsumerChannelMap(topic Topic, ch <-chan *ca.InterContainerMessage) error {
	sc.lockTopicToConsumerChannelMap.Lock()
	defer sc.lockTopicToConsumerChannelMap.Unlock()
	if consumerCh, exist := sc.topicToConsumerChannelMap[topic.Name]; exist {
		// Channel will be closed in the removeChannel method
		consumerCh.channels = removeChannel(consumerCh.channels, ch)
		// If there are no more channels then we can close the consumer itself
		if len(consumerCh.channels) == 0 {
			log.Debugw("closing-consumer", log.Fields{"topic": topic})
			err := consumerCh.consumer.Close()
			delete(sc.topicToConsumerChannelMap, topic.Name)
			return err
		}
		return nil
	}
	log.Warnw("topic-does-not-exist", log.Fields{"topic": topic.Name})
	return errors.New("topic-does-not-exist")
}

func (sc *SaramaClient) clearTopicFromConsumerChannelMap(topic Topic) error {
	sc.lockTopicToConsumerChannelMap.Lock()
	defer sc.lockTopicToConsumerChannelMap.Unlock()
	if consumerCh, exist := sc.topicToConsumerChannelMap[topic.Name]; exist {
		for _, ch := range consumerCh.channels {
			// Channel will be closed in the removeChannel method
			removeChannel(consumerCh.channels, ch)
		}
		err := consumerCh.consumer.Close()
		delete(sc.topicToConsumerChannelMap, topic.Name)
		return err
	}
	log.Warnw("topic-does-not-exist", log.Fields{"topic": topic.Name})
	return errors.New("topic-does-not-exist")
}

func (sc *SaramaClient) clearConsumerChannelMap() error {
	sc.lockTopicToConsumerChannelMap.Lock()
	defer sc.lockTopicToConsumerChannelMap.Unlock()
	var err error
	for topic, consumerCh := range sc.topicToConsumerChannelMap {
		for _, ch := range consumerCh.channels {
			// Channel will be closed in the removeChannel method
			removeChannel(consumerCh.channels, ch)
		}
		err = consumerCh.consumer.Close()
		delete(sc.topicToConsumerChannelMap, topic)
	}
	return err
}

func (sc *SaramaClient) CreateTopic(topic *Topic, numPartition int, repFactor int, retries int) error {
	// This Creates the kafka topic
	// Set broker configuration
	kafkaFullAddr := fmt.Sprintf("%s:%d", sc.KafkaHost, sc.KafkaPort)
	broker := sarama.NewBroker(kafkaFullAddr)

	// Additional configurations. Check sarama doc for more info
	config := sarama.NewConfig()
	config.Version = sarama.V1_0_0_0

	// Open broker connection with configs defined above
	broker.Open(config)

	// check if the connection was OK
	_, err := broker.Connected()
	if err != nil {
		return err
	}

	topicDetail := &sarama.TopicDetail{}
	topicDetail.NumPartitions = int32(numPartition)
	topicDetail.ReplicationFactor = int16(repFactor)
	topicDetail.ConfigEntries = make(map[string]*string)

	topicDetails := make(map[string]*sarama.TopicDetail)
	topicDetails[topic.Name] = topicDetail

	request := sarama.CreateTopicsRequest{
		Timeout:      time.Second * 1,
		TopicDetails: topicDetails,
	}

	for {
		// Send request to Broker
		if response, err := broker.CreateTopics(&request); err != nil {
			if retries == 0 {
				log.Errorw("error-creating-topic", log.Fields{"error": err})
				return err
			} else {
				// If retries is -ve then we will retry indefinitely
				retries--
			}
			log.Debug("retrying-after-a-second-delay")
			time.Sleep(time.Duration(DefaultSleepOnError) * time.Second)
		} else {
			log.Debug("topic-response", log.Fields{"response": response})
			break
		}
	}

	log.Debug("topic-created", log.Fields{"topic": topic, "numPartition": numPartition, "replicationFactor": repFactor})
	return nil
}

func (sc *SaramaClient) createPublisher(retries int) error {
	// This Creates the publisher
	config := sarama.NewConfig()
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.Producer.Flush.Frequency = time.Duration(DefaultFlushFrequency)
	config.Producer.Flush.Messages = DefaultFlushMessages
	config.Producer.Flush.MaxMessages = DefaultFlushMaxmessages
	config.Producer.Return.Errors = DefaultReturnErrors
	config.Producer.Return.Successes = DefaultReturnSuccess
	config.Producer.RequiredAcks = sarama.WaitForAll
	kafkaFullAddr := fmt.Sprintf("%s:%d", sc.KafkaHost, sc.KafkaPort)
	brokers := []string{kafkaFullAddr}

	for {
		producer, err := sarama.NewAsyncProducer(brokers, config)
		if err != nil {
			if retries == 0 {
				log.Errorw("error-starting-publisher", log.Fields{"error": err})
				return err
			} else {
				// If retries is -ve then we will retry indefinitely
				retries--
			}
			log.Info("retrying-after-a-second-delay")
			time.Sleep(time.Duration(DefaultSleepOnError) * time.Second)
		} else {
			sc.producer = producer
			break
		}
	}
	log.Info("Kafka-publisher-created")
	return nil
}

func (sc *SaramaClient) createConsumer(retries int) error {
	config := sarama.NewConfig()
	config.Consumer.Return.Errors = true
	config.Consumer.Fetch.Min = 1
	config.Consumer.MaxWaitTime = time.Duration(DefaultConsumerMaxwait) * time.Millisecond
	config.Consumer.MaxProcessingTime = time.Duration(DefaultMaxProcessingTime) * time.Millisecond
	config.Consumer.Offsets.Initial = sarama.OffsetNewest
	kafkaFullAddr := fmt.Sprintf("%s:%d", sc.KafkaHost, sc.KafkaPort)
	brokers := []string{kafkaFullAddr}

	for {
		consumer, err := sarama.NewConsumer(brokers, config)
		if err != nil {
			if retries == 0 {
				log.Errorw("error-starting-consumer", log.Fields{"error": err})
				return err
			} else {
				// If retries is -ve then we will retry indefinitely
				retries--
			}
			log.Info("retrying-after-a-second-delay")
			time.Sleep(time.Duration(DefaultSleepOnError) * time.Second)
		} else {
			sc.consumer = consumer
			break
		}
	}
	log.Info("Kafka-consumer-created")
	return nil
}

// createGroupConsumer creates a consumer group
func (sc *SaramaClient) createGroupConsumer(topic *Topic, groupId *string, retries int) (*scc.Consumer, error) {
	config := scc.NewConfig()
	config.Group.Mode = scc.ConsumerModeMultiplex
	config.Consumer.Return.Errors = true
	config.Group.Return.Notifications = true
	config.Consumer.MaxWaitTime = time.Duration(DefaultConsumerMaxwait) * time.Millisecond
	config.Consumer.MaxProcessingTime = time.Duration(DefaultMaxProcessingTime) * time.Millisecond
	config.Consumer.Offsets.Initial = sarama.OffsetNewest
	kafkaFullAddr := fmt.Sprintf("%s:%d", sc.KafkaHost, sc.KafkaPort)
	brokers := []string{kafkaFullAddr}

	if groupId == nil {
		g := DefaultGroupName
		groupId = &g
	}
	topics := []string{topic.Name}
	var consumer *scc.Consumer
	var err error

	// Create the topic with default attributes
	// TODO: needs to be revisited
	//sc.CreateTopic(&Topic{Name:topic.Name}, 3, 1, 1)

	if consumer, err = scc.NewConsumer(brokers, *groupId, topics, config); err != nil {
		log.Errorw("create-consumer-failure", log.Fields{"error": err, "topic": topic.Name, "groupId": groupId})
		return nil, err
	}
	log.Debugw("create-consumer-success", log.Fields{"topic": topic.Name, "groupId": groupId})
	//time.Sleep(10*time.Second)
	sc.groupConsumer = consumer
	return consumer, nil
}

// send formats and sends the request onto the kafka messaging bus.
func (sc *SaramaClient) Send(msg interface{}, topic *Topic, keys ...string) {

	// Assert message is a proto message
	var protoMsg proto.Message
	var ok bool
	// ascertain the value interface type is a proto.Message
	if protoMsg, ok = msg.(proto.Message); !ok {
		log.Warnw("message-not-proto-message", log.Fields{"msg": msg})
		return
	}

	//	Create the Sarama producer message
	marshalled, _ := proto.Marshal(protoMsg)
	key := ""
	if len(keys) > 0 {
		key = keys[0] // Only the first key is relevant
	}
	kafkaMsg := &sarama.ProducerMessage{
		Topic: topic.Name,
		Key:   sarama.StringEncoder(key),
		Value: sarama.ByteEncoder(marshalled),
	}

	// Send message to kafka
	sc.producer.Input() <- kafkaMsg
}

// Subscribe registers a caller to a topic.   It returns a channel that the caller can use to receive
// messages from that topic
func (sc *SaramaClient) Subscribe(topic *Topic, retries int) (<-chan *ca.InterContainerMessage, error) {
	log.Debugw("subscribe", log.Fields{"topic": topic.Name})

	// If a consumer already exist for that topic then resuse it
	if consumerCh := sc.getConsumerChannel(*topic); consumerCh != nil {
		log.Debugw("topic-already-subscribed", log.Fields{"topic": topic.Name})
		// Create a channel specific for that consumer and add it to the consumer channel map
		ch := make(chan *ca.InterContainerMessage)
		sc.addChannelToConsumerChannelMap(*topic, ch)
		return ch, nil
	}

	// Register for the topic and set it up
	var consumerListeningChannel chan *ca.InterContainerMessage
	var err error
	if consumerListeningChannel, err = sc.setupConsumerChannel(topic); err != nil {
		log.Warnw("create-consumer-channel-failure", log.Fields{"error": err, "topic": topic.Name})
		return nil, err
	}

	return consumerListeningChannel, nil
}

// dispatchToConsumers sends the intercontainermessage received on a given topic to all subscribers for that
// topic via the unique channel each subsciber received during subscription
func (sc *SaramaClient) dispatchToConsumers(consumerCh *consumerChannels, protoMessage *ca.InterContainerMessage) {
	// Need to go over all channels and publish messages to them - do we need to copy msg?
	sc.lockTopicToConsumerChannelMap.Lock()
	defer sc.lockTopicToConsumerChannelMap.Unlock()
	for _, ch := range consumerCh.channels {
		go func(c chan *ca.InterContainerMessage) {
			c <- protoMessage
		}(ch)
	}
}

func (sc *SaramaClient) consumeMessagesLoop(topic Topic) {
	log.Debugw("starting-consuming-messages", log.Fields{"topic": topic.Name})
	var consumerCh *consumerChannels
	if consumerCh = sc.getConsumerChannel(topic); consumerCh == nil {
		log.Errorw("consumer-not-exist", log.Fields{"topic": topic.Name})
		return
	}
startloop:
	for {
		select {
		case err := <-consumerCh.consumer.Errors():
			log.Warnw("consumer-error", log.Fields{"error": err})
		case msg := <-consumerCh.consumer.Messages():
			//log.Debugw("message-received", log.Fields{"msg": msg, "receivedTopic": msg.Topic})
			// Since the only expected message is a proto intercontainermessage then extract it right away
			// instead of dispatching it to the consumers
			msgBody := msg.Value
			icm := &ca.InterContainerMessage{}
			if err := proto.Unmarshal(msgBody, icm); err != nil {
				log.Warnw("invalid-message", log.Fields{"error": err})
				continue
			}
			go sc.dispatchToConsumers(consumerCh, icm)

			//consumerCh.consumer.MarkOffset(msg, "")
			//// TODO:  Dispatch requests and responses separately
		case <-sc.doneCh:
			log.Infow("received-exit-signal", log.Fields{"topic": topic.Name})
			break startloop
		}
	}
	log.Infow("received-exit-signal-out-of-for-loop", log.Fields{"topic": topic.Name})
}

// setupConsumerChannel creates a consumerChannels object for that topic and add it to the consumerChannels map
// for that topic.  It also starts the routine that listens for messages on that topic.
func (sc *SaramaClient) setupConsumerChannel(topic *Topic) (chan *ca.InterContainerMessage, error) {
	// TODO:  Replace this development partition consumer with a group consumer
	var pConsumer *sarama.PartitionConsumer
	var err error
	if pConsumer, err = sc.CreatePartionConsumer(topic, DefaultMaxRetries); err != nil {
		log.Errorw("creating-partition-consumer-failure", log.Fields{"error": err, "topic": topic.Name})
		return nil, err
	}

	// Create the consumer/channel structure and set the consumer and create a channel on that topic - for now
	// unbuffered to verify race conditions.
	consumerListeningChannel := make(chan *ca.InterContainerMessage)
	cc := &consumerChannels{
		consumer: *pConsumer,
		channels: []chan *ca.InterContainerMessage{consumerListeningChannel},
	}

	// Add the consumer channel to the map
	sc.addTopicToConsumerChannelMap(topic.Name, cc)

	//Start a consumer to listen on that specific topic
	go sc.consumeMessagesLoop(*topic)

	return consumerListeningChannel, nil
}

func (sc *SaramaClient) CreatePartionConsumer(topic *Topic, retries int) (*sarama.PartitionConsumer, error) {
	log.Debugw("creating-partition-consumer", log.Fields{"topic": topic.Name})
	partitionList, err := sc.consumer.Partitions(topic.Name)
	if err != nil {
		log.Warnw("get-partition-failure", log.Fields{"error": err, "topic": topic.Name})
		return nil, err
	}

	log.Debugw("partitions", log.Fields{"topic": topic.Name, "partitionList": partitionList, "first": partitionList[0]})
	// Create a partition consumer for that topic - for now just use one partition
	var pConsumer sarama.PartitionConsumer
	if pConsumer, err = sc.consumer.ConsumePartition(topic.Name, partitionList[0], sarama.OffsetNewest); err != nil {
		log.Warnw("consumer-partition-failure", log.Fields{"error": err, "topic": topic.Name})
		return nil, err
	}
	log.Debugw("partition-consumer-created", log.Fields{"topic": topic.Name})
	return &pConsumer, nil
}

func (sc *SaramaClient) UnSubscribe(topic *Topic, ch <-chan *ca.InterContainerMessage) error {
	log.Debugw("unsubscribing-channel-from-topic", log.Fields{"topic": topic.Name})
	err := sc.removeChannelFromConsumerChannelMap(*topic, ch)
	return err
}

func removeChannel(channels []chan *ca.InterContainerMessage, ch <-chan *ca.InterContainerMessage) []chan *ca.InterContainerMessage {
	var i int
	var channel chan *ca.InterContainerMessage
	for i, channel = range channels {
		if channel == ch {
			channels[len(channels)-1], channels[i] = channels[i], channels[len(channels)-1]
			close(channel)
			return channels[:len(channels)-1]
		}
	}
	return channels
}

func (sc *SaramaClient) DeleteTopic(topic *Topic) error {
	if err := sc.clearTopicFromConsumerChannelMap(*topic); err != nil {
		log.Errorw("failure-clearing-channels", log.Fields{"topic": topic, "error": err})
		return err
	}
	return nil
}