refactor(app): moved all db and log to one intialize spot

This commit is contained in:
2025-08-04 06:54:21 -05:00
parent 486e4fb6b8
commit 0ecbe29ec1
13 changed files with 148 additions and 99 deletions

View File

@@ -2,14 +2,13 @@ package ws
import (
"fmt"
"log"
"sync"
"sync/atomic"
"time"
"github.com/google/uuid"
"github.com/gorilla/websocket"
"lst.net/internal/db"
"gorm.io/gorm"
"lst.net/internal/models"
"lst.net/pkg"
"lst.net/pkg/logger"
@@ -38,7 +37,8 @@ type Client struct {
}
func (c *Client) SaveToDB() {
func (c *Client) SaveToDB(log *logger.CustomLogger, db *gorm.DB) {
// Convert c.Channels (map[string]bool) to map[string]interface{} for JSONB
channels := make(map[string]interface{})
for ch := range c.Channels {
@@ -54,22 +54,27 @@ func (c *Client) SaveToDB() {
LastHeartbeat: time.Now(),
}
if err := db.DB.Create(&clientRecord).Error; err != nil {
log.Println("❌ Error saving client:", err)
if err := db.Create(&clientRecord).Error; err != nil {
log.Error("❌ Error saving client", "websocket", map[string]interface{}{
"error": err,
})
} else {
c.ClientID = clientRecord.ClientID
c.ConnectedAt = clientRecord.ConnectedAt
clientData := fmt.Sprintf("A new client %v, just connected", c.ClientID)
log.Info(clientData, "websocket", map[string]interface{}{})
}
}
func (c *Client) MarkDisconnected() {
logger := logger.New()
clientData := fmt.Sprintf("Client %v just lefts us", c.ClientID)
logger.Info(clientData, "websocket", map[string]interface{}{})
func (c *Client) MarkDisconnected(log *logger.CustomLogger, db *gorm.DB) {
clientData := fmt.Sprintf("Client %v Dicconected", c.ClientID)
log.Info(clientData, "websocket", map[string]interface{}{})
now := time.Now()
res := db.DB.Model(&models.ClientRecord{}).
res := db.Model(&models.ClientRecord{}).
Where("client_id = ?", c.ClientID).
Updates(map[string]interface{}{
"disconnected_at": &now,
@@ -77,13 +82,13 @@ func (c *Client) MarkDisconnected() {
if res.RowsAffected == 0 {
logger.Info("⚠️ No rows updated for client_id", "websocket", map[string]interface{}{
log.Info("⚠️ No rows updated for client_id", "websocket", map[string]interface{}{
"clientID": c.ClientID,
})
}
if res.Error != nil {
logger.Error("❌ Error updating disconnected_at", "websocket", map[string]interface{}{
log.Error("❌ Error updating disconnected_at", "websocket", map[string]interface{}{
"clientID": c.ClientID,
"error": res.Error,
})
@@ -137,28 +142,31 @@ const (
writeWait = 10 * time.Second
)
func (c *Client) StartHeartbeat() {
logger := logger.New()
log.Println("Started hearbeat")
func (c *Client) StartHeartbeat(log *logger.CustomLogger, db *gorm.DB) {
log.Debug("Started hearbeat", "websocket", map[string]interface{}{})
ticker := time.NewTicker(pingPeriod)
defer ticker.Stop()
for {
select {
case <-ticker.C:
if !c.isAlive.Load() { // Correct way to read atomic.Bool
if !c.isAlive.Load() {
return
}
c.Conn.SetWriteDeadline(time.Now().Add(writeWait))
if err := c.Conn.WriteMessage(websocket.PingMessage, nil); err != nil {
log.Printf("Heartbeat failed for %s: %v", c.ClientID, err)
c.Close()
log.Error("Heartbeat failed", "websocket", map[string]interface{}{
"client_id": c.ClientID,
"error": err,
})
c.Close(log, db)
return
}
now := time.Now()
res := db.DB.Model(&models.ClientRecord{}).
res := db.Model(&models.ClientRecord{}).
Where("client_id = ?", c.ClientID).
Updates(map[string]interface{}{
"last_heartbeat": &now,
@@ -166,17 +174,21 @@ func (c *Client) StartHeartbeat() {
if res.RowsAffected == 0 {
logger.Info("⚠️ No rows updated for client_id", "websocket", map[string]interface{}{
log.Info("⚠️ No rows updated for client_id", "websocket", map[string]interface{}{
"clientID": c.ClientID,
})
}
if res.Error != nil {
logger.Error("❌ Error updating disconnected_at", "websocket", map[string]interface{}{
log.Error("❌ Error updating disconnected_at", "websocket", map[string]interface{}{
"clientID": c.ClientID,
"error": res.Error,
})
}
clientStuff := fmt.Sprintf("HeartBeat just done on: %v", c.ClientID)
log.Info(clientStuff, "websocket", map[string]interface{}{
"clientID": c.ClientID,
})
case <-c.done:
return
@@ -184,16 +196,16 @@ func (c *Client) StartHeartbeat() {
}
}
func (c *Client) Close() {
func (c *Client) Close(log *logger.CustomLogger, db *gorm.DB) {
if c.isAlive.CompareAndSwap(true, false) { // Atomic swap
close(c.done)
c.Conn.Close()
// Add any other cleanup here
c.MarkDisconnected()
c.MarkDisconnected(log, db)
}
}
func (c *Client) startServerPings() {
func (c *Client) startServerPings(log *logger.CustomLogger, db *gorm.DB) {
ticker := time.NewTicker(60 * time.Second) // Ping every 30s
defer ticker.Stop()
@@ -202,7 +214,13 @@ func (c *Client) startServerPings() {
case <-ticker.C:
c.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
if err := c.Conn.WriteMessage(websocket.PingMessage, nil); err != nil {
c.Close() // Disconnect if ping fails
log.Error("Server Ping failed", "websocket", map[string]interface{}{
"clientID": c.ClientID,
"error": err,
})
c.Close(log, db)
return
}
case <-c.done:
@@ -219,15 +237,14 @@ func (c *Client) IsActive() bool {
return time.Since(c.lastActive) < 45*time.Second // 1.5x ping interval
}
func (c *Client) updateHeartbeat() {
func (c *Client) updateHeartbeat(log *logger.CustomLogger, db *gorm.DB) {
//fmt.Println("Updating heatbeat")
now := time.Now()
logger := logger.New()
//fmt.Printf("Updating heartbeat for client: %s at %v\n", c.ClientID, now)
//db.DB = db.DB.Debug()
res := db.DB.Model(&models.ClientRecord{}).
res := db.Model(&models.ClientRecord{}).
Where("client_id = ?", c.ClientID).
Updates(map[string]interface{}{
"last_heartbeat": &now, // Explicit format
@@ -235,27 +252,27 @@ func (c *Client) updateHeartbeat() {
//fmt.Printf("Executed SQL: %v\n", db.DB.Statement.SQL.String())
if res.RowsAffected == 0 {
logger.Info("⚠️ No rows updated for client_id", "websocket", map[string]interface{}{
log.Info("⚠️ No rows updated for client_id", "websocket", map[string]interface{}{
"clientID": c.ClientID,
})
}
if res.Error != nil {
logger.Error("❌ Error updating disconnected_at", "websocket", map[string]interface{}{
log.Error("❌ Error updating disconnected_at", "websocket", map[string]interface{}{
"clientID": c.ClientID,
"error": res.Error,
})
}
// 2. Verify DB connection
if db.DB == nil {
logger.Error("DB connection is nil", "websocket", map[string]interface{}{})
if db == nil {
log.Error("DB connection is nil", "websocket", map[string]interface{}{})
return
}
// 3. Test raw SQL execution first
testRes := db.DB.Exec("SELECT 1")
testRes := db.Exec("SELECT 1")
if testRes.Error != nil {
logger.Error("DB ping failed", "websocket", map[string]interface{}{
log.Error("DB ping failed", "websocket", map[string]interface{}{
"error": testRes.Error,
})
return

View File

@@ -2,12 +2,13 @@ package ws
import (
"encoding/json"
"log"
"net/http"
"time"
"github.com/gin-gonic/gin"
"github.com/gorilla/websocket"
"gorm.io/gorm"
"lst.net/pkg/logger"
)
type JoinPayload struct {
@@ -26,11 +27,11 @@ var upgrader = websocket.Upgrader{
EnableCompression: true,
}
func SocketHandler(c *gin.Context, channels map[string]*Channel) {
func SocketHandler(c *gin.Context, channels map[string]*Channel, log *logger.CustomLogger, db *gorm.DB) {
// Upgrade HTTP to WebSocket
conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
if err != nil {
log.Println("WebSocket upgrade failed:", err)
log.Error("WebSocket upgrade failed", "websocket", map[string]interface{}{"error": err})
return
}
//defer conn.Close()
@@ -53,7 +54,7 @@ func SocketHandler(c *gin.Context, channels map[string]*Channel) {
clientsMu.Unlock()
// Save initial connection to DB
client.SaveToDB()
client.SaveToDB(log, db)
// Save initial connection to DB
// if err := client.SaveToDB(); err != nil {
// log.Println("Failed to save client to DB:", err)
@@ -70,12 +71,12 @@ func SocketHandler(c *gin.Context, channels map[string]*Channel) {
now := time.Now()
client.markActive() // Track last pong time
client.lastActive = now
client.updateHeartbeat()
client.updateHeartbeat(log, db)
return nil
})
// Start server-side ping ticker
go client.startServerPings()
go client.startServerPings(log, db)
defer func() {
// Unregister from all channels
@@ -91,11 +92,13 @@ func SocketHandler(c *gin.Context, channels map[string]*Channel) {
clientsMu.Unlock()
// Mark disconnected in DB
client.MarkDisconnected()
client.MarkDisconnected(log, db)
// Close connection
conn.Close()
log.Printf("Client disconnected: %s", client.ClientID)
log.Info("Client disconnected", "websocket", map[string]interface{}{
"client": client.ClientID,
})
}()
// Send welcome message immediately
@@ -104,7 +107,7 @@ func SocketHandler(c *gin.Context, channels map[string]*Channel) {
"message": "Welcome to the WebSocket server. Send subscription request to begin.",
}
if err := conn.WriteJSON(welcomeMsg); err != nil {
log.Println("Failed to send welcome message:", err)
log.Error("Failed to send welcome message", "websocket", map[string]interface{}{"error": err})
return
}
@@ -118,14 +121,14 @@ func SocketHandler(c *gin.Context, channels map[string]*Channel) {
}
}
close(client.Send)
client.MarkDisconnected()
client.MarkDisconnected(log, db)
}()
for {
_, msg, err := conn.ReadMessage()
if err != nil {
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
log.Printf("Client disconnected unexpectedly: %v", err)
log.Error("Client disconnected unexpectedl", "websocket", map[string]interface{}{"error": err})
}
break
}
@@ -173,6 +176,7 @@ func SocketHandler(c *gin.Context, channels map[string]*Channel) {
client.Channels["logServices"] = true
conn.WriteJSON(map[string]string{
"message": "You are now subscribed to the the service channel",
"status": "subscribed",
"channel": "logServices",
})
@@ -194,12 +198,13 @@ func SocketHandler(c *gin.Context, channels map[string]*Channel) {
client.Channels["labels"] = true
// Update DB record
client.SaveToDB()
client.SaveToDB(log, db)
// if err := client.SaveToDB(); err != nil {
// log.Println("Failed to update client labels:", err)
// }
conn.WriteJSON(map[string]interface{}{
"message": "You are now subscribed to the label channel",
"status": "subscribed",
"channel": "labels",
"filters": client.Labels,
@@ -217,7 +222,7 @@ func SocketHandler(c *gin.Context, channels map[string]*Channel) {
// Send messages to client
for message := range client.Send {
if err := conn.WriteMessage(websocket.TextMessage, message); err != nil {
log.Println("Write error:", err)
log.Error("Write erro", "websocket", map[string]interface{}{"error": err})
break
}
}

View File

@@ -26,8 +26,7 @@ import (
"lst.net/pkg/logger"
)
func LogServices(broadcaster chan logger.Message) {
log := logger.New()
func LogServices(broadcaster chan logger.Message, log *logger.CustomLogger) {
log.Info("[LogServices] started - single channel for all logs", "websocket", map[string]interface{}{})
@@ -47,7 +46,7 @@ func LogServices(broadcaster chan logger.Message) {
})
}
fmt.Println("Listening for all logs through single logServices channel...")
log.Info("Listening for all logs through single logServices channel...", "wbsocker", map[string]interface{}{})
for {
select {
case notify := <-listener.Notify:

View File

@@ -4,6 +4,7 @@ import (
"net/http"
"github.com/gin-gonic/gin"
"gorm.io/gorm"
"lst.net/pkg/logger"
)
@@ -11,7 +12,7 @@ var (
broadcaster = make(chan logger.Message)
)
func RegisterSocketRoutes(r *gin.Engine, base_url string) {
func RegisterSocketRoutes(r *gin.Engine, base_url string, log *logger.CustomLogger, db *gorm.DB) {
// Initialize all channels
InitializeChannels()
@@ -19,12 +20,12 @@ func RegisterSocketRoutes(r *gin.Engine, base_url string) {
StartAllChannels()
// Start background services
go LogServices(broadcaster)
go LogServices(broadcaster, log)
go StartBroadcasting(broadcaster, channels)
// WebSocket route
r.GET(base_url+"/ws", func(c *gin.Context) {
SocketHandler(c, channels)
SocketHandler(c, channels, log, db)
})
r.GET(base_url+"/ws/clients", AdminAuthMiddleware(), handleGetClients)

View File

@@ -13,7 +13,7 @@ import (
"lst.net/pkg/logger"
)
func Setup(db *gorm.DB, basePath string) *gin.Engine {
func Setup(db *gorm.DB, basePath string, log *logger.CustomLogger) *gin.Engine {
r := gin.Default()
if os.Getenv("APP_ENV") == "production" {
@@ -45,9 +45,9 @@ func Setup(db *gorm.DB, basePath string) *gin.Engine {
})
// all routes to there respective systems.
ws.RegisterSocketRoutes(r, basePath)
settings.RegisterSettingsRoutes(r, basePath)
servers.RegisterServersRoutes(r, basePath)
ws.RegisterSocketRoutes(r, basePath, log, db)
settings.RegisterSettingsRoutes(r, basePath, log, db)
servers.RegisterServersRoutes(r, basePath, log, db)
r.Any(basePath+"/api", errorApiLoc)

View File

@@ -5,14 +5,14 @@ import (
"strings"
"github.com/gin-gonic/gin"
"lst.net/internal/db"
"gorm.io/gorm"
"lst.net/internal/models"
"lst.net/pkg/logger"
)
func getServers(c *gin.Context) {
log := logger.New()
servers, err := GetServers()
func getServers(c *gin.Context, log *logger.CustomLogger, db *gorm.DB) {
servers, err := GetServers(log, db)
log.Info("Current Settings", "system", map[string]interface{}{
"endpoint": "/api/v1/settings",
"client_ip": c.ClientIP(),
@@ -20,7 +20,7 @@ func getServers(c *gin.Context) {
})
if err != nil {
log := logger.New()
log.Error("Current Settings", "system", map[string]interface{}{
"endpoint": "/api/v1/settings",
"client_ip": c.ClientIP(),
@@ -34,9 +34,9 @@ func getServers(c *gin.Context) {
c.JSON(200, gin.H{"message": "Current settings", "data": servers})
}
func GetServers() ([]map[string]interface{}, error) {
func GetServers(log *logger.CustomLogger, db *gorm.DB) ([]map[string]interface{}, error) {
var servers []models.Servers
res := db.DB.Find(&servers)
res := db.Find(&servers)
if res.Error != nil {
return nil, res.Error

View File

@@ -1,18 +1,19 @@
package servers
import (
"log"
"lst.net/internal/db"
"gorm.io/gorm"
"lst.net/internal/models"
"lst.net/pkg/logger"
)
func NewServer(serverData models.Servers) (string, error) {
func NewServer(serverData models.Servers, log *logger.CustomLogger, db *gorm.DB) (string, error) {
err := db.DB.Create(&serverData).Error
err := db.Create(&serverData).Error
if err != nil {
log.Println("There was an error adding the new server")
log.Error("There was an error adding the new server", "server", map[string]interface{}{
"error": err,
})
return "There was an error adding the new server", err
}

View File

@@ -2,10 +2,12 @@ package servers
import (
"github.com/gin-gonic/gin"
"gorm.io/gorm"
"lst.net/pkg/logger"
)
func RegisterServersRoutes(l *gin.Engine, baseUrl string) {
func RegisterServersRoutes(l *gin.Engine, baseUrl string, log *logger.CustomLogger, db *gorm.DB) {
s := l.Group(baseUrl + "/api/v1")
s.GET("/servers", getServers)
s.GET("/servers", func(c *gin.Context) { getServers(c, log, db) })
}

View File

@@ -4,23 +4,23 @@ import (
"encoding/json"
"github.com/gin-gonic/gin"
"lst.net/internal/db"
"gorm.io/gorm"
"lst.net/pkg/logger"
)
func RegisterSettingsRoutes(l *gin.Engine, baseUrl string) {
func RegisterSettingsRoutes(l *gin.Engine, baseUrl string, log *logger.CustomLogger, db *gorm.DB) {
// seed the db on start up
db.SeedSettings(db.DB)
SeedSettings(db, log)
s := l.Group(baseUrl + "/api/v1")
s.GET("/settings", getSettings)
s.PATCH("/settings/:id", updateSettingById)
s.GET("/settings", func(c *gin.Context) { getSettings(c, log, db) })
s.PATCH("/settings/:id", func(c *gin.Context) { updateSettingById(c, log, db) })
}
func getSettings(c *gin.Context) {
log := logger.New()
configs, err := GetAllSettings(db.DB)
log.Info("Current Settings", "system", map[string]interface{}{
func getSettings(c *gin.Context, log *logger.CustomLogger, db *gorm.DB) {
configs, err := GetAllSettings(db)
log.Info("Current Settings", "settings", map[string]interface{}{
"endpoint": "/api/v1/settings",
"client_ip": c.ClientIP(),
"user_agent": c.Request.UserAgent(),
@@ -28,7 +28,7 @@ func getSettings(c *gin.Context) {
if err != nil {
log := logger.New()
log.Error("Current Settings", "system", map[string]interface{}{
log.Error("Current Settings", "settings", map[string]interface{}{
"endpoint": "/api/v1/settings",
"client_ip": c.ClientIP(),
"user_agent": c.Request.UserAgent(),
@@ -41,13 +41,13 @@ func getSettings(c *gin.Context) {
c.JSON(200, gin.H{"message": "Current settings", "data": configs})
}
func updateSettingById(c *gin.Context) {
log := logger.New()
func updateSettingById(c *gin.Context, log *logger.CustomLogger, db *gorm.DB) {
settingID := c.Param("id")
if settingID == "" {
c.JSON(500, gin.H{"message": "Invalid data"})
log.Error("Invalid data", "system", map[string]interface{}{
log.Error("Invalid data", "settings", map[string]interface{}{
"endpoint": "/api/v1/settings",
"client_ip": c.ClientIP(),
"user_agent": c.Request.UserAgent(),
@@ -63,7 +63,7 @@ func updateSettingById(c *gin.Context) {
if err := decoder.Decode(&setting); err != nil {
c.JSON(400, gin.H{"message": "Invalid request body", "error": err.Error()})
log.Error("Invalid request body", "system", map[string]interface{}{
log.Error("Invalid request body", "settings", map[string]interface{}{
"endpoint": "/api/v1/settings",
"client_ip": c.ClientIP(),
"user_agent": c.Request.UserAgent(),
@@ -72,9 +72,9 @@ func updateSettingById(c *gin.Context) {
return
}
if err := UpdateSetting(db.DB, settingID, setting); err != nil {
if err := UpdateSetting(log, db, settingID, setting); err != nil {
c.JSON(500, gin.H{"message": "Failed to update setting", "error": err.Error()})
log.Error("Failed to update setting", "system", map[string]interface{}{
log.Error("Failed to update setting", "settings", map[string]interface{}{
"endpoint": "/api/v1/settings",
"client_ip": c.ClientIP(),
"user_agent": c.Request.UserAgent(),

View File

@@ -1,10 +1,9 @@
package db
package settings
import (
"log"
"gorm.io/gorm"
"lst.net/internal/models"
"lst.net/pkg/logger"
)
var seedConfigData = []models.Settings{
@@ -44,7 +43,7 @@ var seedConfigData = []models.Settings{
{Name: "stagingReturnLocations", Value: `30125,31523`, Description: "What are the staging location IDs we will use to select from. seperated by commas", AppService: "logistics", Enabled: true},
}
func SeedSettings(db *gorm.DB) error {
func SeedSettings(db *gorm.DB, log *logger.CustomLogger) error {
for _, cfg := range seedConfigData {
var existing models.Settings
@@ -55,7 +54,10 @@ func SeedSettings(db *gorm.DB) error {
if result.Error == gorm.ErrRecordNotFound {
// not here lets add it
if err := db.Create(&cfg).Error; err != nil {
log.Printf("Failed to seed config %s: %v", cfg.Name, err)
log.Error("Failed to seed settings", "settings", map[string]interface{}{
"name": cfg.Name,
"error": err,
})
}
//log.Printf("Seeded new config: %s", cfg.Name)
} else {
@@ -65,13 +67,20 @@ func SeedSettings(db *gorm.DB) error {
} else {
// only update the fields we want to update.
existing.Description = cfg.Description
existing.Name = cfg.Name
existing.AppService = cfg.AppService
if err := db.Save(&existing).Error; err != nil {
log.Printf("Failed to update config %s: %v", cfg.Name, err)
log.Error("Failed to update ettings.", "settings", map[string]interface{}{
"name": cfg.Name,
"error": err,
})
return err
}
//log.Printf("Updated existing config: %s", cfg.Name)
}
}
log.Info("All settings added or updated.", "settings", map[string]interface{}{})
return nil
}

View File

@@ -3,9 +3,10 @@ package settings
import (
"gorm.io/gorm"
"lst.net/internal/models"
"lst.net/pkg/logger"
)
func UpdateSetting(db *gorm.DB, id string, input SettingUpdateInput) error {
func UpdateSetting(log *logger.CustomLogger, db *gorm.DB, id string, input SettingUpdateInput) error {
var cfg models.Settings
if err := db.Where("setting_id =?", id).First(&cfg).Error; err != nil {
return err
@@ -30,5 +31,18 @@ func UpdateSetting(db *gorm.DB, id string, input SettingUpdateInput) error {
return nil // nothing to update
}
return db.Model(&cfg).Updates(updates).Error
settingUpdate := db.Model(&cfg).Updates(updates)
if settingUpdate.Error != nil {
log.Error("There was an error updating the setting", "settings", map[string]interface{}{
"error": settingUpdate.Error,
})
return settingUpdate.Error
}
log.Info("The setting was just updated", "settings", map[string]interface{}{
"name": settingUpdate.Name,
})
return nil
}