package test

import (
	"github.com/google/uuid"
	"github.com/sirupsen/logrus"
	"github.com/testcontainers/testcontainers-go"
	"github.com/testcontainers/testcontainers-go/wait"
	"kafka-stream-operator/src/internal/gateways/kafka/publisher"
	"log"
	"strings"
	"testing"
	"time"
)

var kafkaPublisher 	*publisher.Publisher
var bootstrapURL = "localhost:9092"
var compose *testcontainers.LocalDockerCompose
var testTopic = "test-in"

func TestMain(m *testing.M) {
	defer teardown()
    setup()
	m.Run()
}

func setup(){
	log.Println("Setup")
    compose = testcontainers.NewLocalDockerCompose([]string{"../../docker-compose.yml"}
		, strings.ToLower(uuid.New().String()))
	compose.WaitForService("kafka_1",wait.ForLog("Ready to serve as the new controller with epoch 1"))
	compose.WithCommand([]string{"up", "-d"}).Invoke()
    logrus.Info("Containers started! ")

	var err error
	kafkaPublisher, err = publisher.NewPublisher(bootstrapURL, testTopic)
	if err !=nil {
		logrus.Errorf("Failed to start publisher: %v", err)
	}

}

func teardown(){
	log.Println("Teardown")
   compose.Down()
   time.Sleep(1 * time.Second)
}


func sendMessage(message string) error{

   err := kafkaPublisher.Publish("", message)
	if err !=nil {
		logrus.Errorf("Failed to send mesaage: %v", message)
       return err
	}
   return nil
}


func TestSuccess(t *testing.T) {
   log.Println("testSuccess running")

   message := `{"log":"testing log for user test", "kubernetes":{"host":"docker-desktop"}}`

   err := sendMessage(message)
   if err != nil {
       t.Errorf("Failed to send message: %v", err)
       t.Fail()
   }
   err = sendMessage(message)
   if err != nil {
       t.Errorf("Failed to send message: %v", err)
       t.Fail()
   }

}

version: "2"

services:
  zookeeper:
    image: bitnami/zookeeper:3.7
    ports:
      - "2181:2181"
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes

  kafka:
    image: docker.io/bitnami/kafka:3
    ports:
      - "9092:9092"
    environment:
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_CFG_LISTENERS=INTERNAL://:9093,EXTERNAL://:9092
      - KAFKA_CFG_ADVERTISED_LISTENERS=INTERNAL://:9093,EXTERNAL://localhost:9092
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=INTERNAL
    depends_on:
      - zookeeper

  kowl:
    image: rsmnarts/kowl:latest
    restart: on-failure
    ports:
      - "8080:8080"
    environment:
      - KAFKA_BROKERS=kafka:9093
    depends_on:
      - kafka