package broker

import (
	"context"
	"fmt"
	"time"

	"github.com/segmentio/kafka-go"
	"gitlab.dcso.lolcat/LABS/styx/models"
)

func SetUpKafkaConnecter() (*kafka.Conn, error) {
	topic := "styx"
	partition := 0
	conn, err := kafka.DialLeader(context.Background(), "tcp", "localhost:9092", topic, partition)
	if err != nil {
		return nil, err
	}
	conn.SetDeadline(time.Now().Add(10 * time.Second))

	return conn, nil
}

func ProduceEvent(conn *kafka.Conn, message string) error {
	// SetWriteDeadline sets the deadline for future Write calls and any
	// currently-blocked Write call. Even if write times out, it may return n > 0,
	// indicating that some of the data was successfully written. A zero value for t
	// means Write will not time out.
	conn.SetWriteDeadline(time.Now().Add(10 * time.Second))

	_, err := conn.WriteMessages(
		kafka.Message{Value: []byte("one!")},
		kafka.Message{Value: []byte("two!")},
		kafka.Message{Value: []byte("three!")},
	)
	if err != nil {
		return nil
	}

	return nil
}

func SendEventToKafka(conn *kafka.Conn, node models.Node) error {
	conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
	_, err := conn.WriteMessages(kafka.Message{Value: []byte(fmt.Sprintf("%v", node))})
	if err != nil {
		panic(err)
	}

	return nil
}

func ReadEventFromKafka() {
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:   []string{"localhost:9092"},
		Topic:     "styx",
		Partition: 0,
		MinBytes:  10e3,
		MaxBytes:  10e6,
	})
	r.SetOffset(42)
	defer r.Close()

	for {
		m, err := r.ReadMessage(context.Background())
		if err != nil {
			panic(err)
		}

		fmt.Printf("message at offset %d:%s = %s\n", m.Offset, string(m.Key), string(m.Value))
	}

}