working Gateway for incoming Mod/Bot ws-conn's; simulated Mod client via sim.go

ws/handlers.go

@@ -1 +1,107 @@
 package ws
+
+import (
+	"encoding/json"
+	"net/http"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+func (wsg *WebsocketGateway) handlePush(w http.ResponseWriter, r *http.Request) {
+	conn, err := wsg.validateAndUpgradeConnection(w, r)
+	if err != nil {
+		return
+	}
+
+	if wsg.bodySizeBytes > 0 {
+		conn.SetReadLimit(wsg.bodySizeBytes)
+	} else {
+		conn.SetReadLimit(1 << 20) // sensible default 1MiB
+	}
+
+	_ = conn.SetReadDeadline(time.Now().Add(60 * time.Second))
+	conn.SetPongHandler(func(appData string) error {
+		_ = conn.SetReadDeadline(time.Now().Add(60 * time.Second))
+		return nil
+	})
+
+	typ, data, err := conn.ReadMessage()
+	if err != nil {
+		wsg.sendWebsocketError(conn, "Internal Server Error", 500)
+		wsg.logger.Error("Failed to read handshake.", "remote", conn.RemoteAddr().String(), "err", err)
+		return
+	}
+
+	if typ != websocket.TextMessage && typ != websocket.BinaryMessage {
+		wsg.sendWebsocketError(conn, "First message must be a handshake.", 400)
+		wsg.logger.Warn("Invalid handshake message type.", "remote", conn.RemoteAddr().String())
+		return
+	}
+
+	var handshake Handshake
+	if err := json.Unmarshal(data, &handshake); err != nil {
+		wsg.sendWebsocketError(conn, "Malformed handshake.", 400)
+		wsg.logger.Warn("Malformed handshake.", "remote", conn.RemoteAddr().String(), "err", err)
+		return
+	}
+
+	meta := connectionMetaData{connectionType: handshake.Type}
+
+	switch handshake.Type {
+	case "mod":
+		var mhs ModHandshake
+
+		if err := json.Unmarshal(handshake.Data, &mhs); err != nil {
+			wsg.sendWebsocketError(conn, "Malformed mod handshake.", 400)
+			wsg.logger.Warn("Malformed mod handshake.", "remote", conn.RemoteAddr().String(), "err", err)
+			return
+		}
+
+		meta.id = mhs.ServerID
+
+		if err = wsg.sendWebsocketResponse(conn, GatewayAck{Status: "connected", Type: "mod"}); err != nil {
+			return
+		}
+
+		wsg.registerConn(conn, meta)
+		wsg.logger.Info("Mod connected via Websocket.", "remote", conn.RemoteAddr().String(), "server_id", mhs.ServerID)
+
+		go wsg.modReadLoop(conn, meta) // replace with external handler mayhaps
+
+	case "bot":
+		var bhs BotHandshake
+
+		if err := json.Unmarshal(handshake.Data, &bhs); err != nil {
+			wsg.sendWebsocketError(conn, "Malformed bot handshake.", 400)
+			wsg.logger.Warn("Malformed bot handshake.", "remote", conn.RemoteAddr().String(), "err", err)
+			return
+		}
+
+		meta.id = bhs.BotID
+
+		if err = wsg.sendWebsocketResponse(conn, GatewayAck{Status: "connected", Type: "bot"}); err != nil {
+			return
+		}
+
+		wsg.registerConn(conn, meta)
+		wsg.logger.Info("Bot connected via Websocket.", "remote", conn.RemoteAddr().String(), "bot_id", bhs.BotID)
+
+		go wsg.botReadLoop(conn, meta) // replace with external handler mayhaps
+
+	default:
+		wsg.sendWebsocketError(conn, "Unknown handshake.", 400)
+		wsg.logger.Warn("Unknown connection type.", "remote", conn.RemoteAddr().String(), "type", handshake.Type)
+		return
+	}
+}
+
+func (wsg *WebsocketGateway) handleReady(w http.ResponseWriter, r *http.Request) {}
+
+func (wsg *WebsocketGateway) handleHealth(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(200)
+	_ = json.NewEncoder(w).Encode(map[string]interface{}{"status": "healthy"})
+}
+
+func (wsg *WebsocketGateway) handleRegister(w http.ResponseWriter, r *http.Request) {}

ws/temp.go

@@ -0,0 +1,70 @@
+package ws
+
+import (
+	"context"
+	"encoding/json"
+	"log/slog"
+	"time"
+)
+
+type LoggingModHandler struct {
+	logger *slog.Logger
+}
+
+type LoggingBotHandler struct {
+	logger *slog.Logger
+}
+
+func NewLoggingModHandler(logger *slog.Logger) *LoggingModHandler {
+	return &LoggingModHandler{logger: logger}
+}
+
+func NewLoggingBotHandler(logger *slog.Logger) *LoggingBotHandler {
+	return &LoggingBotHandler{logger: logger}
+}
+
+func (h *LoggingModHandler) Handle(ctx context.Context, msg GatewayModMessageIn) error {
+	// For now, just log and pretend it's being forwarded
+	// TODO: Look up channel_id from database using server
+	// TODO: Forward to bot connection(s)
+
+	fwd := GatewayModMessageOut{
+		Type:        "mod",
+		ChannelID:   "TODO", // will come from database lookup
+		Author:      msg.Author,
+		Content:     msg.Content,
+		Meta:        msg.Meta,
+		Ts:          msg.Ts,
+		ReceivedAt:  msg.ReceivedAt,
+		ForwardedAt: time.Now().UTC(),
+	}
+
+	b, _ := json.Marshal(fwd)
+	h.logger.Info("received mod message", "msg_id", msg.MsgID, "server", msg.Server, "Author", msg.Author.Name, "content", msg.Content)
+	h.logger.Debug("forwarding mod message", "msg_id", msg.MsgID, "server", msg.Server, "payload", string(b))
+
+	return nil
+}
+
+func (h *LoggingBotHandler) Handle(ctx context.Context, msg GatewayBotMessageIn) error {
+	// For now, just log and pretend it's being forwarded
+	// TODO: Look up server_id from database using channel_id
+	// TODO: Forward to mod connection(s)
+
+	fwd := GatewayBotMessageOut{
+		Type:        "bot",
+		ChannelID:   msg.ChannelID,
+		Author:      msg.Author,
+		Content:     msg.Content,
+		Meta:        msg.Meta,
+		Ts:          msg.Ts,
+		ReceivedAt:  msg.ReceivedAt,
+		ForwardedAt: time.Now().UTC(),
+	}
+
+	b, _ := json.Marshal(fwd)
+	h.logger.Info("received bot message", "msg_id", msg.MsgID, "channel", msg.ChannelID, "author", msg.Author, "content", msg.Content)
+	h.logger.Debug("forwarding bot message", "msg_id", msg.MsgID, "channel", msg.ChannelID, "payload", string(b))
+
+	return nil
+}

ws/validate.go

@@ -12,8 +12,8 @@ func (m *GatewayModMessageIn) Validate() error {
 	if strings.TrimSpace(m.Server) == "" {
 		return errors.New("server missing")
 	}
-	if strings.TrimSpace(m.User.ID) == "" {
-		return errors.New("user.id missing")
+	if strings.TrimSpace(m.Author.ID) == "" {
+		return errors.New("author.id missing")
 	}
 	if strings.TrimSpace(m.Content) == "" {
 		return errors.New("content missing")
@@ -28,7 +28,7 @@ func (m *GatewayBotMessageIn) Validate() error {
 	if strings.TrimSpace(m.ChannelID) == "" {
 		return errors.New("channel_id missing")
 	}
-	if strings.TrimSpace(m.Author) == "" {
+	if strings.TrimSpace(m.Author.ID) == "" {
 		return errors.New("author missing")
 	}
 	if strings.TrimSpace(m.Content) == "" {

ws/websocket.go

@@ -69,106 +69,6 @@ func (wsg *WebsocketGateway) Serve(ctx context.Context, listenAddr string) error
 
 //
 
-func (wsg *WebsocketGateway) handlePush(w http.ResponseWriter, r *http.Request) {
-	conn, err := wsg.validateAndUpgradeConnection(w, r)
-	if err != nil {
-		return
-	}
-
-	if wsg.bodySizeBytes > 0 {
-		conn.SetReadLimit(wsg.bodySizeBytes)
-	} else {
-		conn.SetReadLimit(1 << 20) // sensible default 1MiB
-	}
-
-	_ = conn.SetReadDeadline(time.Now().Add(60 * time.Second))
-	conn.SetPongHandler(func(appData string) error {
-		_ = conn.SetReadDeadline(time.Now().Add(60 * time.Second))
-		return nil
-	})
-
-	typ, data, err := conn.ReadMessage()
-	if err != nil {
-		wsg.sendWebsocketError(conn, "Internal Server Error", 500)
-		wsg.logger.Error("Failed to read handshake.", "remote", conn.RemoteAddr().String(), "err", err)
-		return
-	}
-
-	if typ != websocket.TextMessage && typ != websocket.BinaryMessage {
-		wsg.sendWebsocketError(conn, "First message must be a handshake.", 400)
-		wsg.logger.Warn("Invalid handshake message type.", "remote", conn.RemoteAddr().String())
-		return
-	}
-
-	var handshake Handshake
-	if err := json.Unmarshal(data, &handshake); err != nil {
-		wsg.sendWebsocketError(conn, "Malformed handshake.", 400)
-		wsg.logger.Warn("Malformed handshake.", "remote", conn.RemoteAddr().String(), "err", err)
-		return
-	}
-
-	meta := connectionMetaData{connectionType: handshake.Type}
-
-	switch handshake.Type {
-	case "mod":
-		var mhs ModHandshake
-
-		if err := json.Unmarshal(handshake.Data, &mhs); err != nil {
-			wsg.sendWebsocketError(conn, "Malformed mod handshake.", 400)
-			wsg.logger.Warn("Malformed mod handshake.", "remote", conn.RemoteAddr().String(), "err", err)
-			return
-		}
-
-		meta.id = mhs.ServerID
-
-		if err = wsg.sendWebsocketResponse(conn, GatewayAck{Status: "connected", Type: "mod"}); err != nil {
-			return
-		}
-
-		wsg.registerConn(conn, meta)
-		wsg.logger.Info("Mod connected via Websocket.", "remote", conn.RemoteAddr().String(), "server_id", mhs.ServerID)
-
-		go wsg.modReadLoop(conn, meta) // replace with external handler mayhaps
-
-	case "bot":
-		var bhs BotHandshake
-
-		if err := json.Unmarshal(handshake.Data, &bhs); err != nil {
-			wsg.sendWebsocketError(conn, "Malformed bot handshake.", 400)
-			wsg.logger.Warn("Malformed bot handshake.", "remote", conn.RemoteAddr().String(), "err", err)
-			return
-		}
-
-		meta.id = bhs.BotID
-
-		if err = wsg.sendWebsocketResponse(conn, GatewayAck{Status: "connected", Type: "bot"}); err != nil {
-			return
-		}
-
-		wsg.registerConn(conn, meta)
-		wsg.logger.Info("Bot connected via Websocket.", "remote", conn.RemoteAddr().String(), "bot_id", bhs.BotID)
-
-		go wsg.botReadLoop(conn, meta) // replace with external handler mayhaps
-
-	default:
-		wsg.sendWebsocketError(conn, "Unknown handshake.", 400)
-		wsg.logger.Warn("Unknown connection type.", "remote", conn.RemoteAddr().String(), "type", handshake.Type)
-		return
-	}
-}
-
-func (wsg *WebsocketGateway) handleReady(w http.ResponseWriter, r *http.Request) {}
-
-func (wsg *WebsocketGateway) handleHealth(w http.ResponseWriter, r *http.Request) {
-	w.Header().Set("Content-Type", "application/json")
-	w.WriteHeader(200)
-	_ = json.NewEncoder(w).Encode(map[string]interface{}{"status": "healthy"})
-}
-
-func (wsg *WebsocketGateway) handleRegister(w http.ResponseWriter, r *http.Request) {}
-
-//
-
 func (wsg *WebsocketGateway) modReadLoop(c *websocket.Conn, meta connectionMetaData) {
 	defer func() {
 		wsg.unregisterConn(c)
@@ -179,6 +79,18 @@ func (wsg *WebsocketGateway) modReadLoop(c *websocket.Conn, meta connectionMetaD
 	pingTicker := time.NewTicker(30 * time.Second)
 	defer pingTicker.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 {
 		typ, data, err := c.ReadMessage()
 		if err != nil {
@@ -216,17 +128,6 @@ func (wsg *WebsocketGateway) modReadLoop(c *websocket.Conn, meta connectionMetaD
 		}
 
 		_ = writeJSONSafe(c, map[string]string{"status": "ok"})
-
-		// Handle pings
-		select {
-		case <-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
-			}
-		default:
-		}
 	}
 }
 
@@ -240,6 +141,18 @@ func (wsg *WebsocketGateway) botReadLoop(c *websocket.Conn, meta connectionMetaD
 	pingTicker := time.NewTicker(30 * time.Second)
 	defer pingTicker.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", "bot_id", meta.id, "err", err)
+				return
+			}
+			wsg.logger.Debug("sent ping to bot", "bot_id", meta.id)
+		}
+	}()
+
 	for {
 		typ, data, err := c.ReadMessage()
 		if err != nil {
@@ -277,16 +190,5 @@ func (wsg *WebsocketGateway) botReadLoop(c *websocket.Conn, meta connectionMetaD
 		}
 
 		_ = writeJSONSafe(c, map[string]string{"status": "ok"})
-
-		// Handle pings
-		select {
-		case <-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)
-				return
-			}
-		default:
-		}
 	}
 }