Homestead/HomesteadGateway
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

eod commit

Browse Source
This commit is contained in:
Overlord
2025-12-01 22:30:36 +01:00
parent 2372da942a
commit 8f7db6256b
9 changed files with 190 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
ws/websocket.go
View File

@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"fmt"
"homestead/homestead_gateway/util/cache"
"homestead/homestead_gateway/util/config"
"log/slog"
"net"
@@ -15,11 +16,16 @@ import (
"github.com/gorilla/websocket"
)
func NewWsGateway(cfg config.GatewayConfig, logger *slog.Logger, closefn func() error, modH ModHandler, botH BotHandler) *WebsocketGateway {
func NewWebsocketGateway(cfg config.GatewayConfig, logger *slog.Logger, closefn func() error, modH ModHandler, botH BotHandler) *WebsocketGateway {
return &WebsocketGateway{
logger: logger,
closeFn: closefn,
apiKey: cfg.Websocket,
logger: logger,
closeFn: closefn,
modHandler: modH,
botHandler: botH,
port: cfg.HttpPort,
apiKey: cfg.Websocket,
cache: cache.NewCache(),
conns: cache.NewConnectionCache(),
upgrader: websocket.Upgrader{
ReadBufferSize: 1024,
WriteBufferSize: 1024,
@@ -27,11 +33,7 @@ func NewWsGateway(cfg config.GatewayConfig, logger *slog.Logger, closefn func()
return true // local by default; change for production
},
},
conns: make(map[*websocket.Conn]connectionMetaData),
bodySizeBytes: int64(cfg.BodySize) * 1024 * 1024,
port: cfg.HttpPort,
modHandler: modH,
botHandler: botH,
}
}
@@ -69,35 +71,29 @@ func (wsg *WebsocketGateway) Serve(ctx context.Context, listenAddr string) error
//
func (wsg *WebsocketGateway) modReadLoop(c *websocket.Conn, meta connectionMetaData) {
func (wsg *WebsocketGateway) modReadLoop(conn *websocket.Conn, meta cache.ConnectionMetaData) {
defer func() {
wsg.unregisterConn(c)
_ = c.Close()
wsg.logger.Info("mod disconnected", "server_id", meta.id, "remote", c.RemoteAddr().String())
wsg.unregisterConn(conn, meta)
wsg.logger.Info("Mod-Client disconnected.", "remote", conn.RemoteAddr().String(), "server_id", meta.ID)
}()
pingTicker := time.NewTicker(30 * time.Second)
defer pingTicker.Stop()
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
// Send pings in a separate goroutine
go func() {
for range pingTicker.C {
_ = c.SetWriteDeadline(time.Now().Add(5 * time.Second))
if err := c.WriteMessage(websocket.PingMessage, nil); err != nil {
wsg.logger.Debug("write ping failed", "server_id", meta.id, "err", err)
return
}
wsg.logger.Debug("sent ping to mod", "server_id", meta.id)
for range ticker.C {
wsg.sendWebsocketPing(conn)
}
}()
for {
typ, data, err := c.ReadMessage()
typ, data, err := conn.ReadMessage()
if err != nil {
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
wsg.logger.Warn("unexpected mod close", "server_id", meta.id, "err", err)
wsg.logger.Warn("unexpected mod close", "server_id", meta.ID, "err", err)
} else {
wsg.logger.Debug("mod read error", "server_id", meta.id, "err", err)
wsg.logger.Debug("mod read error", "server_id", meta.ID, "err", err)
}
return
}
@@ -108,34 +104,33 @@ func (wsg *WebsocketGateway) modReadLoop(c *websocket.Conn, meta connectionMetaD
var msg GatewayModMessageIn
if err := json.Unmarshal(data, &msg); err != nil {
_ = writeJSONSafe(c, map[string]string{"error": "invalid json: " + err.Error()})
wsg.logger.Warn("invalid json from mod", "server_id", meta.id, "remote", c.RemoteAddr().String(), "err", err)
_ = writeJSONSafe(conn, map[string]string{"error": "invalid json: " + err.Error()})
wsg.logger.Warn("invalid json from mod", "server_id", meta.ID, "remote", conn.RemoteAddr().String(), "err", err)
continue
}
msg.ReceivedAt = time.Now().UTC()
if err := msg.Validate(); err != nil {
_ = writeJSONSafe(c, map[string]string{"error": err.Error()})
wsg.logger.Warn("mod message validation failed", "server_id", meta.id, "remote", c.RemoteAddr().String(), "err", err)
_ = writeJSONSafe(conn, map[string]string{"error": err.Error()})
wsg.logger.Warn("mod message validation failed", "server_id", meta.ID, "remote", conn.RemoteAddr().String(), "err", err)
continue
}
// Handle the message (forward to bot, enrich, etc.)
if err := wsg.modHandler.Handle(context.Background(), msg); err != nil {
_ = writeJSONSafe(c, map[string]string{"error": "handler error: " + err.Error()})
wsg.logger.Error("mod handler error", "server_id", meta.id, "err", err)
if err := wsg.modHandler.Handle(conn, msg); err != nil {
_ = writeJSONSafe(conn, map[string]string{"error": "handler error: " + err.Error()})
wsg.logger.Error("mod handler error", "server_id", meta.ID, "err", err)
continue
}
_ = writeJSONSafe(c, map[string]string{"status": "ok"})
_ = writeJSONSafe(conn, map[string]string{"status": "completed"}) // or "queued"
}
}
func (wsg *WebsocketGateway) botReadLoop(c *websocket.Conn, meta connectionMetaData) {
func (wsg *WebsocketGateway) botReadLoop(conn *websocket.Conn, meta cache.ConnectionMetaData) {
defer func() {
wsg.unregisterConn(c)
_ = c.Close()
wsg.logger.Info("bot disconnected", "bot_id", meta.id, "remote", c.RemoteAddr().String())
wsg.unregisterConn(conn, meta)
wsg.logger.Info("bot disconnected", "bot_id", meta.ID, "remote", conn.RemoteAddr().String())
}()
pingTicker := time.NewTicker(30 * time.Second)
@@ -144,22 +139,22 @@ func (wsg *WebsocketGateway) botReadLoop(c *websocket.Conn, meta connectionMetaD
// Send pings in a separate goroutine
go func() {
for range pingTicker.C {
_ = c.SetWriteDeadline(time.Now().Add(5 * time.Second))
if err := c.WriteMessage(websocket.PingMessage, nil); err != nil {
wsg.logger.Debug("write ping failed", "bot_id", meta.id, "err", err)
_ = conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil {
wsg.logger.Debug("write ping failed", "bot_id", meta.ID, "err", err)
return
}
wsg.logger.Debug("sent ping to bot", "bot_id", meta.id)
wsg.logger.Debug("sent ping to bot", "bot_id", meta.ID)
}
}()
for {
typ, data, err := c.ReadMessage()
typ, data, err := conn.ReadMessage()
if err != nil {
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
wsg.logger.Warn("unexpected bot close", "bot_id", meta.id, "err", err)
wsg.logger.Warn("unexpected bot close", "bot_id", meta.ID, "err", err)
} else {
wsg.logger.Debug("bot read error", "bot_id", meta.id, "err", err)
wsg.logger.Debug("bot read error", "bot_id", meta.ID, "err", err)
}
return
}
@@ -170,25 +165,25 @@ func (wsg *WebsocketGateway) botReadLoop(c *websocket.Conn, meta connectionMetaD
var msg GatewayBotMessageIn
if err := json.Unmarshal(data, &msg); err != nil {
_ = writeJSONSafe(c, map[string]string{"error": "invalid json: " + err.Error()})
wsg.logger.Warn("invalid json from bot", "bot_id", meta.id, "remote", c.RemoteAddr().String(), "err", err)
_ = writeJSONSafe(conn, map[string]string{"error": "invalid json: " + err.Error()})
wsg.logger.Warn("invalid json from bot", "bot_id", meta.ID, "remote", conn.RemoteAddr().String(), "err", err)
continue
}
msg.ReceivedAt = time.Now().UTC()
if err := msg.Validate(); err != nil {
_ = writeJSONSafe(c, map[string]string{"error": err.Error()})
wsg.logger.Warn("bot message validation failed", "bot_id", meta.id, "remote", c.RemoteAddr().String(), "err", err)
_ = writeJSONSafe(conn, map[string]string{"error": err.Error()})
wsg.logger.Warn("bot message validation failed", "bot_id", meta.ID, "remote", conn.RemoteAddr().String(), "err", err)
continue
}
// Handle the message (forward to mod, enrich, etc.)
if err := wsg.botHandler.Handle(context.Background(), msg); err != nil {
_ = writeJSONSafe(c, map[string]string{"error": "handler error: " + err.Error()})
wsg.logger.Error("bot handler error", "bot_id", meta.id, "err", err)
if err := wsg.botHandler.Handle(conn, msg); err != nil {
_ = writeJSONSafe(conn, map[string]string{"error": "handler error: " + err.Error()})
wsg.logger.Error("bot handler error", "bot_id", meta.ID, "err", err)
continue
}
_ = writeJSONSafe(c, map[string]string{"status": "ok"})
_ = writeJSONSafe(conn, map[string]string{"status": "ok"})
}
}