From aaeb9a5c871dc660c07e45cd0aab12439d7ee0bb Mon Sep 17 00:00:00 2001
From: Michael Matloka <michal@matloka.com>
Date: Mon, 10 Jun 2024 10:45:08 +0200
Subject: [PATCH] Don't connect to Kafka over SSL in dev

---
 kafka.go |  4 ++--
 main.go  | 10 ++++++++--
 2 files changed, 10 insertions(+), 4 deletions(-)

diff --git a/kafka.go b/kafka.go
index 690fbe7..4988e78 100644
--- a/kafka.go
+++ b/kafka.go
@@ -35,13 +35,13 @@ type KafkaConsumer struct {
 	statsChan    chan PostHogEvent
 }
 
-func NewKafkaConsumer(brokers string, groupID string, topic string, geolocator *GeoLocator, outgoingChan chan PostHogEvent, statsChan chan PostHogEvent) (*KafkaConsumer, error) {
+func NewKafkaConsumer(brokers string, securityProtocol string, groupID string, topic string, geolocator *GeoLocator, outgoingChan chan PostHogEvent, statsChan chan PostHogEvent) (*KafkaConsumer, error) {
 	config := &kafka.ConfigMap{
 		"bootstrap.servers":  brokers,
 		"group.id":           groupID,
 		"auto.offset.reset":  "latest",
 		"enable.auto.commit": false,
-		"security.protocol":  "SSL",
+		"security.protocol":  securityProtocol,
 	}
 
 	consumer, err := kafka.NewConsumer(config)
diff --git a/main.go b/main.go
index 4049d5e..c059ce2 100644
--- a/main.go
+++ b/main.go
@@ -19,6 +19,8 @@ import (
 func main() {
 	loadConfigs()
 
+	isProd := viper.GetBool("prod")
+
 	mmdb := viper.GetString("mmdb.path")
 	if mmdb == "" {
 		log.Fatal("mmdb.path must be set")
@@ -52,7 +54,11 @@ func main() {
 
 	go teamStats.keepStats(statsChan)
 
-	consumer, err := NewKafkaConsumer(brokers, groupID, topic, geolocator, phEventChan, statsChan)
+	kafkaSecurityProtocol := "SSL"
+	if !isProd {
+		kafkaSecurityProtocol = "PLAINTEXT"
+	}
+	consumer, err := NewKafkaConsumer(brokers, kafkaSecurityProtocol, groupID, topic, geolocator, phEventChan, statsChan)
 	if err != nil {
 		log.Fatalf("Failed to create Kafka consumer: %v", err)
 	}
@@ -258,7 +264,7 @@ func main() {
 		}
 	})
 
-	if !viper.GetBool("prod") {
+	if !isProd {
 		e.Logger.Fatal(e.Start(":8080"))
 	} else {
 		e.Logger.Fatal(e.StartAutoTLS(":443"))