go super app.

.env

KAFKA_BROKER=localhost:9092
KAFKA_TOPIC=email-notifications

docker-compose.yaml

version: '3'
services:
 # zookeeper
  zookeeperapp: 
    image: 'wurstmeister/zookeeper:latest' 
    ports: 
      - '2181:2181' 
    environment: 
      - ALLOW_ANONYMOUS_LOGIN=yes
    volumes: 
      - ./volumes/zookeeper:/var/lib/zookeeper/data
  # kafka
  kafkaapp: 
    image: 'wurstmeister/kafka:latest' 
    ports: 
      - '9092:9092' 
    environment: 
      - KAFKA_BROKER_ID=1 
      - KAFKA_LISTENERS=PLAINTEXT://:9092 
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092 
      - KAFKA_ZOOKEEPER_CONNECT=zookeeperapp:2181 
      - ALLOW_PLAINTEXT_LISTENER=yes
      # auto create topic set true
      - KAFKA_AUTO_CREATE_TOPICS_ENABLE=true
    volumes:
      - ./volumes/kafka:/var/lib/kafka/data
    depends_on: 
      - zookeeperapp

go.mod

module super-app

go 1.24.9

require (
	github.com/joho/godotenv v1.5.1
	github.com/labstack/echo/v4 v4.15.0
	github.com/lmittmann/tint v1.1.3
	github.com/segmentio/kafka-go v0.4.50
)

require (
	github.com/klauspost/compress v1.15.9 // indirect
	github.com/labstack/gommon v0.4.2 // indirect
	github.com/mattn/go-colorable v0.1.14 // indirect
	github.com/mattn/go-isatty v0.0.20 // indirect
	github.com/pierrec/lz4/v4 v4.1.15 // indirect
	github.com/valyala/bytebufferpool v1.0.0 // indirect
	github.com/valyala/fasttemplate v1.2.2 // indirect
	golang.org/x/crypto v0.46.0 // indirect
	golang.org/x/net v0.48.0 // indirect
	golang.org/x/sys v0.39.0 // indirect
	golang.org/x/text v0.32.0 // indirect
)

main.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"log/slog"
	"net/http"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	"github.com/joho/godotenv"
	"github.com/labstack/echo/v4"
	"github.com/lmittmann/tint"
	"github.com/segmentio/kafka-go"
)

// EmailMessage defines the structure of the message expected from Kafka
type EmailMessage struct {
	Email   string `json:"email"`
	Content string `json:"content"`
}

func main() {
	// Configure slog with Tint handler for colorful output
	logger := slog.New(tint.NewHandler(os.Stderr, &tint.Options{
		Level:      slog.LevelDebug,
		TimeFormat: time.Kitchen,
	}))
	slog.SetDefault(logger)

	// Load environment variables from .env file
	if err := godotenv.Load("sample.env"); err != nil {
		slog.Warn("No .env file found, using defaults or system env vars")
	}

	// Create a WaitGroup to manage the goroutines
	var wg sync.WaitGroup
	// We have three services: Echo server, Kafka consumer, and Kafka producer (test)
	wg.Add(3)

	// Create a channel to signal graceful shutdown to goroutines
	shutdownCh := make(chan struct{})

	// Start Echo Server
	go startEchoServer(&wg, shutdownCh, logger.With("component", "echo_server"))

	// Start Kafka Consumer
	go consumeKafkaMessage(&wg, shutdownCh, logger.With("component", "kafka_consumer"))

	// Start Kafka Producer (Test)
	go produceTestMessages(&wg, shutdownCh, logger.With("component", "kafka_producer"))

	// Listen for OS signals (keyboard interrupts, termination)
	// This runs in the main thread
	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM)

	// Block until a signal is received
	sig := <-sigCh
	slog.Info("Received signal", "signal", sig)
	slog.Info("Initiating graceful shutdown...")

	// Signal goroutines to stop
	close(shutdownCh)

	// Wait for all goroutines to finish
	wg.Wait()

	slog.Info("bye bye")
}

func startEchoServer(wg *sync.WaitGroup, shutdownCh <-chan struct{}, logger *slog.Logger) {
	defer wg.Done()

	e := echo.New()
	e.HideBanner = true

	// Define the /hello endpoint
	e.GET("/hello", func(c echo.Context) error {
		return c.String(http.StatusOK, "Hello, World!")
	})

	// Start server in a separate goroutine so we can listen on shutdownCh
	go func() {
		if err := e.Start(":8080"); err != nil && err != http.ErrServerClosed {
			logger.Error("shutting down the server", "error", err)
			os.Exit(1) // Fatal error
		}
	}()

	logger.Info("Echo server started on :8080")

	// Wait for shutdown signal
	<-shutdownCh
	logger.Info("Echo server stopping...")

	// Create a context with timeout for graceful shutdown
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	if err := e.Shutdown(ctx); err != nil {
		logger.Error("Error shutting down Echo server", "error", err)
	}
	logger.Info("Echo server stopped")
}

func consumeKafkaMessage(wg *sync.WaitGroup, shutdownCh <-chan struct{}, logger *slog.Logger) {
	defer wg.Done()

	// Kafka Reader configuration (Placeholder values for now)
	// User said they will set up connection later, so we use minimal config
	// preventing immediate panic if env vars are missing, or just assume defaults.
	broker := os.Getenv("KAFKA_BROKER")
	if broker == "" {
		broker = "localhost:9092"
	}
	topic := os.Getenv("KAFKA_TOPIC")
	if topic == "" {
		topic = "email-notifications"
	}
	groupID := "email-consumer-group"

	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers: []string{broker},
		Topic:   topic,
		GroupID: groupID,
	})

	fmt.Printf("Kafka consumer started (Broker: %s, Topic: %s)\n", broker, topic)

	// We need to consume messages and also listen for shutdown.
	// specific implementation for cancellation with kafka-go using context.

	// Create a context that is canceled when shutdownCh is closed
	ctx, cancel := context.WithCancel(context.Background())

	// Monitor shutdownCh to cancel the context
	go func() {
		<-shutdownCh
		fmt.Println("Kafka consumer stopping...")
		cancel()
	}()

	for {
		// ReadMessage blocks execution. We use ReadMessage with context to allow cancellation.
		m, err := reader.ReadMessage(ctx)
		if err != nil {
			// If context was canceled, it means we are shutting down
			if ctx.Err() != nil {
				break
			}
			log.Printf("Error reading message: %v\n", err)
			// Backoff or continue based on error type, for skeleton we just sleep briefly
			time.Sleep(1 * time.Second)
			continue
		}

		// Process the message
		var emailMsg EmailMessage
		if err := json.Unmarshal(m.Value, &emailMsg); err != nil {
			log.Printf("Error unmarshalling message: %v\n", err)
			continue
		}

		fmt.Printf("Received email task: To: %s, Content: %s\n", emailMsg.Email, emailMsg.Content)

		// Here actual email sending logic would go
	}

	// Close the reader
	if err := reader.Close(); err != nil {
		log.Printf("Error closing Kafka reader: %v\n", err)
	} else {
		fmt.Println("Kafka reader closed")
	}
	fmt.Println("Kafka consumer stopped")
}

func produceTestMessages(wg *sync.WaitGroup, shutdownCh <-chan struct{}, logger *slog.Logger) {
	defer wg.Done()

	broker := os.Getenv("KAFKA_BROKER")
	if broker == "" {
		broker = "localhost:9092"
	}
	topic := os.Getenv("KAFKA_TOPIC")
	if topic == "" {
		topic = "email-notifications"
	}

	// Check if topic exists, if not create it (auto-create might be enabled on broker but good to be safe/explicit if needed)
	// For simplify, we rely on auto-creation or pre-existence.

	writer := &kafka.Writer{
		Addr:     kafka.TCP(broker),
		Topic:    topic,
		Balancer: &kafka.LeastBytes{},
	}

	logger.Info("Kafka test producer started", "broker", broker, "topic", topic)

	ticker := time.NewTicker(3 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-shutdownCh:
			logger.Info("Kafka producer stopping...")
			if err := writer.Close(); err != nil {
				logger.Error("Error closing Kafka writer", "error", err)
			}
			logger.Info("Kafka producer stopped")
			return
		case t := <-ticker.C:
			msg := EmailMessage{
				Email:   "test@example.com",
				Content: fmt.Sprintf("Test message generated at %v", t.Format(time.RFC3339)),
			}
			jsonMsg, err := json.Marshal(msg)
			if err != nil {
				logger.Error("Error marshalling test message", "error", err)
				continue
			}

			err = writer.WriteMessages(context.Background(),
				kafka.Message{
					Key:   []byte(fmt.Sprintf("key-%d", t.Unix())),
					Value: jsonMsg,
				},
			)
			if err != nil {
				logger.Error("Error sending test message", "error", err)
			} else {
				logger.Info("Produced test message", "message_content", string(jsonMsg))
			}
		}
	}
}