res_levis_backend/internal/app/server/app.go

package server

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"os"
	"sync"

	"github.com/AFASystems/presence/internal/pkg/apiclient"
	"github.com/AFASystems/presence/internal/pkg/common/appcontext"
	"github.com/AFASystems/presence/internal/pkg/config"
	"github.com/AFASystems/presence/internal/pkg/database"
	"github.com/AFASystems/presence/internal/pkg/kafkaclient"
	"github.com/AFASystems/presence/internal/pkg/logger"
	"github.com/AFASystems/presence/internal/pkg/model"
	"github.com/AFASystems/presence/internal/pkg/service"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

// ServerApp holds dependencies and state for the server service.
type ServerApp struct {
	Cfg              *config.Config
	DB               *gorm.DB
	KafkaManager     *kafkaclient.KafkaManager
	AppState         *appcontext.AppState
	ChLoc            chan model.HTTPLocation
	ChEvents         chan appcontext.BeaconEvent
	ChHealthLocation chan appcontext.LocationHealth
	ChHealthDecoder  chan appcontext.DecoderHealth
	ChHealthBridge   chan appcontext.BridgeHealth
	ctx              context.Context
	Server           *http.Server
	Cleanup          func()
	wg               sync.WaitGroup
}

// New creates a ServerApp: loads config, creates logger, connects DB, creates Kafka manager and writers.
// Caller must call Init(ctx) then Run(ctx) then Shutdown().
func New(cfg *config.Config) (*ServerApp, error) {
	srvLogger, cleanup := logger.CreateLogger("server.log")
	slog.SetDefault(srvLogger)

	db, err := database.Connect(cfg)
	if err != nil {
		cleanup()
		return nil, fmt.Errorf("database: %w", err)
	}

	appState := appcontext.NewAppState()
	kafkaManager := kafkaclient.InitKafkaManager()

	writerTopics := []string{"apibeacons", "alert", "mqtt", "settings", "parser"}
	kafkaManager.PopulateKafkaManager(cfg.KafkaURL, "", writerTopics)
	slog.Info("Kafka writers initialized", "topics", writerTopics)

	return &ServerApp{
		Cfg:          cfg,
		DB:           db,
		KafkaManager: kafkaManager,
		AppState:     appState,
		Cleanup:      cleanup,
	}, nil
}

// Init loads config from file, seeds DB, runs UpdateDB, adds Kafka readers and starts consumers.
func (a *ServerApp) Init(ctx context.Context) error {
	a.ctx = ctx

	configFile, err := os.Open(a.Cfg.ConfigPath)
	if err != nil {
		return fmt.Errorf("config file: %w", err)
	}
	defer configFile.Close()

	b, err := io.ReadAll(configFile)
	if err != nil {
		return fmt.Errorf("read config: %w", err)
	}

	var configs []model.Config
	if err := json.Unmarshal(b, &configs); err != nil {
		return fmt.Errorf("unmarshal config: %w", err)
	}

	for _, c := range configs {
		a.DB.Clauses(clause.OnConflict{DoNothing: true}).Create(&c)
	}
	a.DB.Find(&configs)
	for _, c := range configs {
		kp := model.KafkaParser{ID: "add", Config: c}
		if err := service.SendParserConfig(kp, a.KafkaManager.GetWriter("parser"), ctx); err != nil {
			slog.Error("sending parser config to kafka", "err", err, "name", c.Name)
		}
	}

	if err := apiclient.UpdateDB(a.DB, ctx, a.Cfg, a.KafkaManager.GetWriter("apibeacons"), a.AppState); err != nil {
		slog.Error("UpdateDB", "err", err)
	}

	readerTopics := []string{"locevents", "alertbeacons", "healthlocation", "healthdecoder", "healthbridge"}
	a.KafkaManager.PopulateKafkaManager(a.Cfg.KafkaURL, "server", readerTopics)
	slog.Info("Kafka readers initialized", "topics", readerTopics)

	a.ChLoc = make(chan model.HTTPLocation, config.SMALL_CHANNEL_SIZE)
	a.ChEvents = make(chan appcontext.BeaconEvent, config.MEDIUM_CHANNEL_SIZE)
	a.ChHealthLocation = make(chan appcontext.LocationHealth, config.SMALL_CHANNEL_SIZE)
	a.ChHealthDecoder = make(chan appcontext.DecoderHealth, config.SMALL_CHANNEL_SIZE)
	a.ChHealthBridge = make(chan appcontext.BridgeHealth, config.SMALL_CHANNEL_SIZE)

	a.wg.Add(5)
	go kafkaclient.Consume(a.KafkaManager.GetReader("locevents"), a.ChLoc, ctx, &a.wg)
	go kafkaclient.Consume(a.KafkaManager.GetReader("alertbeacons"), a.ChEvents, ctx, &a.wg)
	go kafkaclient.Consume(a.KafkaManager.GetReader("healthlocation"), a.ChHealthLocation, ctx, &a.wg)
	go kafkaclient.Consume(a.KafkaManager.GetReader("healthdecoder"), a.ChHealthDecoder, ctx, &a.wg)
	go kafkaclient.Consume(a.KafkaManager.GetReader("healthbridge"), a.ChHealthBridge, ctx, &a.wg)

	a.Server = &http.Server{
		Addr:    a.Cfg.HTTPAddr,
		Handler: a.RegisterRoutes(),
	}
	return nil
}

// Run starts the HTTP server and runs the event loop until ctx is cancelled.
func (a *ServerApp) Run(ctx context.Context) {
	go func() {
		if err := a.Server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			slog.Error("HTTP server", "err", err)
		}
	}()
	RunEventLoop(ctx, a)
}

// Shutdown stops the HTTP server, waits for consumers, and cleans up Kafka and logger.
func (a *ServerApp) Shutdown() {
	if a.Server != nil {
		if err := a.Server.Shutdown(context.Background()); err != nil {
			slog.Error("server shutdown", "err", err)
		}
		slog.Info("HTTP server stopped")
	}
	a.wg.Wait()
	slog.Info("Kafka consumers stopped")
	a.KafkaManager.CleanKafkaReaders()
	a.KafkaManager.CleanKafkaWriters()
	if a.Cleanup != nil {
		a.Cleanup()
	}
	slog.Info("server shutdown complete")
}