package mock

import (
	"fmt"
	"sync"
)

// KafkaMock mock for Kafka utility
type KafkaMock struct {
	lock            sync.Mutex
	err             error
	produce         string
	ProduceMessages map[string]map[string]interface{}
}

// Produce mocks produce method
func (k *KafkaMock) Produce(topic string, key string, payload interface{}, headers map[string][]byte) error {
	k.lock.Lock()
	defer k.lock.Unlock()

	if k.err != nil {
		return k.err
	}

	k.produce = fmt.Sprintf("%s %v", topic, payload)
	k.addMessage(topic, key, payload)

	return nil
}

func (k *KafkaMock) ProduceBinary(topic string, key string, data []byte, header map[string][]byte) error {
	if k.err != nil {
		return k.err
	}

	k.produce = fmt.Sprintf("%s %v", topic, data)
	k.addMessage(topic, key, data)

	return nil
}

func (k *KafkaMock) ProduceBinaryToChannel(topic string, key string, payload []byte) error {
	if k.err != nil {
		return k.err
	}

	k.produce = fmt.Sprintf("%s %v", topic, payload)
	k.addMessage(topic, key, payload)

	return nil
}

/*
func (k *KafkaMock) ProduceBatch(topic string, key string, payload [][]byte) error {
	if k.err != nil {
		return k.err
	}

	for _, item := range payload {
		k.addMessage(topic, key, item)
	}

	return nil
}*/

func (k *KafkaMock) ProduceToChannel(topic string, key string, payload interface{}) error {
	k.addMessage(topic, key, payload)

	return nil
}

func (k *KafkaMock) ProduceSignalBatch(topic string, key string, payload interface{}) error {
	if k.err != nil {
		return k.err
	}

	k.produce = fmt.Sprintf("%s %v", topic, payload)
	k.addMessage(topic, key, payload)

	return nil
}

func (k *KafkaMock) ReadEvents() {
	// nothing
}

// Close mocks close method
func (k *KafkaMock) Close() {
	// nothing
}

// SetError to set error for mock Kafka utility
func (k *KafkaMock) SetError(err error) {
	k.err = err
}

func (k *KafkaMock) addMessage(topic string, id string, payload interface{}) {
	if k.ProduceMessages == nil {
		k.ProduceMessages = map[string]map[string]interface{}{}
	}

	if k.ProduceMessages[topic] == nil {
		k.ProduceMessages[topic] = map[string]interface{}{}
	}

	k.ProduceMessages[topic][id] = payload
}

func (k *KafkaMock) Reset() {
	k.ProduceMessages = map[string]map[string]interface{}{}
}

func (k *KafkaMock) Len() int {
	return 0
}

func (k *KafkaMock) Flush(timeoutMs int) int {
	return 0
}

// GetProduce to set error for mock Kafka utility
func (k *KafkaMock) GetProduce() string {
	return k.produce
}

// GetKafkaMock returns mock Kafka utilty
func GetKafkaMock(err error) *KafkaMock {
	return &KafkaMock{
		err: err,
	}
}