res_levis_backend/cmd/server/main.go

package main

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

	"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/controller"
	"github.com/AFASystems/presence/internal/pkg/database"
	"github.com/AFASystems/presence/internal/pkg/kafkaclient"
	"github.com/AFASystems/presence/internal/pkg/model"
	"github.com/AFASystems/presence/internal/pkg/service"
	"github.com/gorilla/handlers"
	"github.com/gorilla/mux"
	"github.com/gorilla/websocket"
	"github.com/segmentio/kafka-go"
	"gorm.io/gorm"
)

var upgrader = websocket.Upgrader{
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
}

var _ io.Writer = (*os.File)(nil)
var wg sync.WaitGroup

func main() {
	cfg := config.Load()
	appState := appcontext.NewAppState()

	// Create log file
	logFile, err := os.OpenFile("server.log", os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
	if err != nil {
		log.Fatalf("Failed to open log file: %v\n", err)
	}
	// shell and log file multiwriter
	w := io.MultiWriter(os.Stderr, logFile)
	logger := slog.New(slog.NewJSONHandler(w, nil))
	slog.SetDefault(logger)

	// define context
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
	defer stop()

	db, err := database.Connect(cfg)
	if err != nil {
		log.Fatalf("Failed to open database connection: %v\n", err)
	}

	headersOk := handlers.AllowedHeaders([]string{"X-Requested-With"})
	originsOk := handlers.AllowedOrigins([]string{"*"})
	methodsOk := handlers.AllowedMethods([]string{"GET", "HEAD", "POST", "PUT", "DELETE", "OPTIONS"})

	writer := appState.AddKafkaWriter(cfg.KafkaURL, "apibeacons")
	settingsWriter := appState.AddKafkaWriter(cfg.KafkaURL, "settings")
	alertWriter := appState.AddKafkaWriter(cfg.KafkaURL, "alert")
	parserWriter := appState.AddKafkaWriter(cfg.KafkaURL, "parser")
	mqttWriter := appState.AddKafkaWriter(cfg.KafkaURL, "mqtt")
	slog.Info("Kafka writers topics: apibeacons, settings initialized")

	configFile, err := os.Open("/app/cmd/server/config.json")
	if err != nil {
		panic(err)
	}

	b, _ := io.ReadAll(configFile)

	var configs []model.Config
	json.Unmarshal(b, &configs)

	for _, config := range configs {
		// persist read configs in database
		db.Create(&config)
	}

	db.Find(&configs)
	for _, config := range configs {
		kp := model.KafkaParser{
			ID:     "add",
			Config: config,
		}

		if err := service.SendParserConfig(kp, parserWriter, ctx); err != nil {
			fmt.Printf("Unable to send parser config to kafka broker %v\n", err)
		}
	}

	if err := apiclient.UpdateDB(db, ctx, cfg, writer, appState); err != nil {
		fmt.Printf("Error in getting token: %v\n", err)
	}

	locationReader := appState.AddKafkaReader(cfg.KafkaURL, "locevents", "gid-loc-server")
	alertsReader := appState.AddKafkaReader(cfg.KafkaURL, "alertbeacons", "gid-alert-serv")
	slog.Info("Kafka readers topics: locevents, alertbeacons initialized")

	chLoc := make(chan model.HTTPLocation, 200)
	chEvents := make(chan model.BeaconEvent, 500)

	wg.Add(2)
	go kafkaclient.Consume(locationReader, chLoc, ctx, &wg)
	go kafkaclient.Consume(alertsReader, chEvents, ctx, &wg)

	r := mux.NewRouter()

	r.HandleFunc("/reslevis/getGateways", controller.GatewayListController(db)).Methods("GET")
	r.HandleFunc("/reslevis/postGateway", controller.GatewayAddController(db)).Methods("POST")
	r.HandleFunc("/reslevis/removeGateway/{id}", controller.GatewayDeleteController(db)).Methods("DELETE")
	r.HandleFunc("/reslevis/updateGateway/{id}", controller.GatewayUpdateController(db)).Methods("PUT")

	r.HandleFunc("/reslevis/getZones", controller.ZoneListController(db)).Methods("GET")
	r.HandleFunc("/reslevis/postZone", controller.ZoneAddController(db)).Methods("POST")
	r.HandleFunc("/reslevis/removeZone/{id}", controller.ZoneDeleteController(db)).Methods("DELETE")
	r.HandleFunc("/reslevis/updateZone", controller.ZoneUpdateController(db)).Methods("PUT")

	r.HandleFunc("/reslevis/getTrackerZones", controller.TrackerZoneListController(db)).Methods("GET")
	r.HandleFunc("/reslevis/postTrackerZone", controller.TrackerZoneAddController(db)).Methods("POST")
	r.HandleFunc("/reslevis/removeTrackerZone/{id}", controller.TrackerZoneDeleteController(db)).Methods("DELETE")
	r.HandleFunc("/reslevis/updateTrackerZone", controller.TrackerZoneUpdateController(db)).Methods("PUT")

	r.HandleFunc("/reslevis/getTrackers", controller.TrackerList(db)).Methods("GET")
	r.HandleFunc("/reslevis/postTracker", controller.TrackerAdd(db, writer, ctx)).Methods("POST")
	r.HandleFunc("/reslevis/removeTracker/{id}", controller.TrackerDelete(db, writer, ctx)).Methods("DELETE")
	r.HandleFunc("/reslevis/updateTracker", controller.TrackerUpdate(db)).Methods("PUT")

	r.HandleFunc("/configs/beacons", controller.ParserListController(db)).Methods("GET")
	r.HandleFunc("/configs/beacons", controller.ParserAddController(db, parserWriter, ctx)).Methods("POST")
	r.HandleFunc("/configs/beacons/{id}", controller.ParserUpdateController(db, parserWriter, ctx)).Methods("PUT")
	r.HandleFunc("/configs/beacons/{id}", controller.ParserDeleteController(db, parserWriter, ctx)).Methods("DELETE")

	r.HandleFunc("/reslevis/settings", controller.SettingsUpdateController(db, settingsWriter, ctx)).Methods("PATCH")
	r.HandleFunc("/reslevis/settings", controller.SettingsListController(db)).Methods("GET")

	beaconTicker := time.NewTicker(2 * time.Second)

	wsHandler := http.HandlerFunc(serveWs(db, ctx))
	restApiHandler := handlers.CORS(originsOk, headersOk, methodsOk)(r)
	mainHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if strings.HasPrefix(r.URL.Path, "/api/beacons/ws") {
			wsHandler.ServeHTTP(w, r)
			return
		}

		restApiHandler.ServeHTTP(w, r)
	})

	server := http.Server{
		Addr:    cfg.HTTPAddr,
		Handler: mainHandler,
	}

	go server.ListenAndServe()

eventLoop:
	for {
		select {
		case <-ctx.Done():
			break eventLoop
		case msg := <-chLoc:
			service.LocationToBeaconService(msg, db, alertWriter, ctx)
		case msg := <-chEvents:
			fmt.Printf("event: %+v\n", msg)
			id := msg.ID
			if err := db.First(&model.Tracker{}, "id = ?", id).Error; err != nil {
				fmt.Printf("Decoder event for untracked beacon: %s\n", id)
				continue
			}

			if err := db.Updates(&model.Tracker{ID: id, Battery: msg.Battery}).Error; err != nil {
				fmt.Printf("Error in saving decoder event for beacon: %s\n", id)
				continue
			}
		case <-beaconTicker.C:
			var list []model.Tracker
			db.Find(&list)
			eMsg, err := json.Marshal(list)
			if err != nil {
				fmt.Printf("Error in marshaling trackers list: %v\n", err)
				continue
			}

			msg := kafka.Message{
				Value: eMsg,
			}

			mqttWriter.WriteMessages(ctx, msg)
		}
	}

	if err := server.Shutdown(context.Background()); err != nil {
		eMsg := fmt.Sprintf("could not shutdown: %v\n", err)
		slog.Error(eMsg)
	}

	slog.Info("API SERVER: \n")
	slog.Warn("broken out of the main event loop and HTTP server shutdown\n")
	wg.Wait()

	slog.Info("All go routines have stopped, Beggining to close Kafka connections\n")
	appState.CleanKafkaReaders()
	appState.CleanKafkaWriters()

	slog.Info("All kafka clients shutdown, starting shutdown of valkey client")
	slog.Info("API server shutting down")
	logFile.Close()
}

func serveWs(db *gorm.DB, ctx context.Context) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		ws, err := upgrader.Upgrade(w, r, nil)
		if err != nil {
			if _, ok := err.(websocket.HandshakeError); !ok {
				eMsg := fmt.Sprintf("could not upgrade ws connection: %v\n", err)
				slog.Error(eMsg)
			}
			return
		}
		wg.Add(2)
		go writer(ws, db, ctx)
		go reader(ws, ctx)
	}
}

func writer(ws *websocket.Conn, db *gorm.DB, ctx context.Context) {
	pingTicker := time.NewTicker((60 * 9) / 10 * time.Second)
	beaconTicker := time.NewTicker(2 * time.Second)
	defer func() {
		pingTicker.Stop()
		beaconTicker.Stop()
		ws.Close()
		wg.Done()
	}()
	for {
		select {
		case <-ctx.Done():
			slog.Info("WebSocket writer received shutdown signal.")
			ws.SetWriteDeadline(time.Now().Add(10 * time.Second))
			ws.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
			return
		case <-beaconTicker.C:
			var list []model.Tracker
			db.Find(&list)
			js, err := json.Marshal(list)
			if err != nil {
				js = []byte("error")
			}

			ws.SetWriteDeadline(time.Now().Add(10 * time.Second))
			if err := ws.WriteMessage(websocket.TextMessage, js); err != nil {
				return
			}
		case <-pingTicker.C:
			ws.SetWriteDeadline(time.Now().Add(10 * time.Second))
			if err := ws.WriteMessage(websocket.PingMessage, []byte{}); err != nil {
				return
			}
		}
	}
}

func reader(ws *websocket.Conn, ctx context.Context) {
	defer func() {
		ws.Close()
		wg.Done()
	}()
	ws.SetReadLimit(512)
	ws.SetReadDeadline(time.Now().Add((60 * 9) / 10 * time.Second))
	ws.SetPongHandler(func(string) error { ws.SetReadDeadline(time.Now().Add((60 * 9) / 10 * time.Second)); return nil })
	for {
		select {
		case <-ctx.Done():
			slog.Info("closing ws reader")
			return
		default:
			_, _, err := ws.ReadMessage()
			if err != nil {
				return
			}
		}
	}
}