Compare commits

...

2 Commits

11 changed files with 612 additions and 443 deletions

View File

@@ -1,3 +1,15 @@
# HomesteadGateway # HomesteadGateway
Gateway between multiple HomesteadRelay's and the HomesteadToGo Bot. Gateway between multiple HomesteadRelay's and the HomesteadToGo Bot.
## dev notes
perhaps drop database, instead
```
Mod -> websocket /register { server_id, channel_id } // grabbed from mod config
Bot -> websocket /ready { channel_id } // ready if Mod with fitting channel_id has called /register
Gateway -> memory cache (server_id -> channel_id; channel_id -> server_id) // mem enough
Mod/Bot -> websocket /ws { ... } -> Bot/Mod // sync
```

15
main.go
View File

@@ -2,10 +2,8 @@ package main
import ( import (
"flag" "flag"
"fmt"
"homestead/homestead_gateway/controller" "homestead/homestead_gateway/controller"
"homestead/homestead_gateway/util/config" "homestead/homestead_gateway/util/config"
"homestead/homestead_gateway/util/logger"
) )
func main() { func main() {
@@ -15,19 +13,6 @@ func main() {
panic(err) panic(err)
} }
l, c, e := logger.New("yomama", cfg.Log)
if e != nil {
panic(e)
}
defer c()
l.Debug("debug")
l.Info("info")
l.Warn("warn")
l.Error("error")
panic(fmt.Sprintf("%+v", cfg.Log))
ctrl := controller.NewGatewayController(*cfg) ctrl := controller.NewGatewayController(*cfg)
err = ctrl.Run() err = ctrl.Run()
if err != nil { if err != nil {

308
sim.go
View File

@@ -2,15 +2,23 @@ package main
import ( import (
"encoding/json" "encoding/json"
"fmt"
"log" "log"
"net/url" "net/url"
"os"
"os/signal"
"syscall"
"time" "time"
"github.com/gorilla/websocket" "github.com/gorilla/websocket"
) )
type MessageEnvelope struct { const (
gatewayURL = "ws://localhost:3333/push"
apiKey = "gateway"
serverID = "test-server-001"
)
type Handshake struct {
Type string `json:"type"` Type string `json:"type"`
Data json.RawMessage `json:"data"` Data json.RawMessage `json:"data"`
} }
@@ -19,173 +27,163 @@ type ModHandshake struct {
ServerID string `json:"server_id"` ServerID string `json:"server_id"`
} }
type BotHandshake struct { type GatewayAck struct {
BotID string `json:"bot_id"` Status string `json:"status"`
Type string `json:"type"`
} }
type User struct { type MinecraftUser struct {
ID string `json:"id"` ID string `json:"id"`
Name string `json:"name"` Name string `json:"name"`
} }
type ModMessage struct { type Destination struct {
ChannelID string `json:"channel_id"`
}
type GatewayModMessageIn struct {
MsgID string `json:"msg_id"` MsgID string `json:"msg_id"`
Server string `json:"server"` Server string `json:"server"`
User User `json:"user"` Destination Destination `json:"destination"`
Author MinecraftUser `json:"author"`
Content string `json:"content"` Content string `json:"content"`
Meta map[string]interface{} `json:"meta,omitempty"` Meta map[string]interface{} `json:"meta,omitempty"`
Ts string `json:"ts,omitempty"` Ts string `json:"ts,omitempty"`
} }
type BotMessage struct {
MsgID string `json:"msg_id"`
ChannelID string `json:"channel_id"`
Author string `json:"author"`
Content string `json:"content"`
Meta map[string]interface{} `json:"meta,omitempty"`
Ts string `json:"ts,omitempty"`
}
func simulateMod() {
u := url.URL{Scheme: "ws", Host: "localhost:3333", Path: "/ws"}
u.RawQuery = "api_key=gateway"
conn, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
if err != nil {
log.Fatalf("mod dial error: %v", err)
}
defer conn.Close()
fmt.Println("[MOD] Connected")
// Send handshake
handshake := ModHandshake{ServerID: "survival_server"}
hsData, _ := json.Marshal(handshake)
envelope := MessageEnvelope{
Type: "mod",
Data: hsData,
}
envData, _ := json.Marshal(envelope)
if err := conn.WriteMessage(websocket.TextMessage, envData); err != nil {
log.Fatalf("mod handshake write error: %v", err)
}
fmt.Println("[MOD] Sent handshake")
// Read handshake response
_, resp, err := conn.ReadMessage()
if err != nil {
log.Fatalf("mod handshake response error: %v", err)
}
fmt.Printf("[MOD] Handshake response: %s\n", string(resp))
// Send a few messages
for i := 1; i <= 3; i++ {
msg := ModMessage{
MsgID: fmt.Sprintf("mod_msg_%d", i),
Server: "survival_server",
User: User{ID: "player_123", Name: "Steve"},
Content: fmt.Sprintf("Message %d from Minecraft!", i),
Meta: map[string]interface{}{
"coordinates": map[string]int{"x": 100 + i*10, "y": 64, "z": 200},
},
Ts: time.Now().UTC().Format(time.RFC3339),
}
msgData, _ := json.Marshal(msg)
if err := conn.WriteMessage(websocket.TextMessage, msgData); err != nil {
log.Fatalf("mod message write error: %v", err)
}
fmt.Printf("[MOD] Sent message: %s\n", msg.Content)
// Read response
_, resp, err := conn.ReadMessage()
if err != nil {
log.Fatalf("mod response error: %v", err)
}
fmt.Printf("[MOD] Response: %s\n", string(resp))
time.Sleep(1 * time.Second)
}
fmt.Println("[MOD] Closing connection")
}
func simulateBot() {
time.Sleep(2 * time.Second) // Let mod connect first
u := url.URL{Scheme: "ws", Host: "localhost:3333", Path: "/ws"}
u.RawQuery = "api_key=gateway"
conn, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
if err != nil {
log.Fatalf("bot dial error: %v", err)
}
defer conn.Close()
fmt.Println("[BOT] Connected")
// Send handshake
handshake := BotHandshake{BotID: "discord_bot_1"}
hsData, _ := json.Marshal(handshake)
envelope := MessageEnvelope{
Type: "bot",
Data: hsData,
}
envData, _ := json.Marshal(envelope)
if err := conn.WriteMessage(websocket.TextMessage, envData); err != nil {
log.Fatalf("bot handshake write error: %v", err)
}
fmt.Println("[BOT] Sent handshake")
// Read handshake response
_, resp, err := conn.ReadMessage()
if err != nil {
log.Fatalf("bot handshake response error: %v", err)
}
fmt.Printf("[BOT] Handshake response: %s\n", string(resp))
// Send a few messages
for i := 1; i <= 3; i++ {
msg := BotMessage{
MsgID: fmt.Sprintf("bot_msg_%d", i),
ChannelID: "987654321",
Author: "DiscordUser#1234",
Content: fmt.Sprintf("Message %d from Discord!", i),
Meta: map[string]interface{}{
"reactions": []string{"👍", "❤️"},
},
Ts: time.Now().UTC().Format(time.RFC3339),
}
msgData, _ := json.Marshal(msg)
if err := conn.WriteMessage(websocket.TextMessage, msgData); err != nil {
log.Fatalf("bot message write error: %v", err)
}
fmt.Printf("[BOT] Sent message: %s\n", msg.Content)
// Read response
_, resp, err := conn.ReadMessage()
if err != nil {
log.Fatalf("bot response error: %v", err)
}
fmt.Printf("[BOT] Response: %s\n", string(resp))
time.Sleep(1 * time.Second)
}
fmt.Println("[BOT] Closing connection")
}
func main() { func main() {
fmt.Println("Starting WebSocket client simulator...") interrupt := make(chan os.Signal, 1)
fmt.Println("Connecting to ws://localhost:3333/ws with api_key=test_key") signal.Notify(interrupt, os.Interrupt, syscall.SIGTERM)
go simulateMod() // Build WebSocket URL with API key
go simulateBot() u, err := url.Parse(gatewayURL)
if err != nil {
log.Fatalf("Failed to parse URL: %v", err)
}
q := u.Query()
q.Set("api_key", apiKey)
u.RawQuery = q.Encode()
// Let them run log.Printf("Connecting to %s", u.String())
time.Sleep(15 * time.Second)
fmt.Println("Simulator finished") // Connect to WebSocket
conn, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
if err != nil {
log.Fatalf("Failed to connect: %v", err)
}
defer conn.Close()
log.Println("Connected to gateway")
// Set up ping handler - respond to pings from server
conn.SetPingHandler(func(appData string) error {
log.Println("Received ping from server, sending pong")
err := conn.WriteControl(websocket.PongMessage, []byte(appData), time.Now().Add(5*time.Second))
if err != nil {
log.Printf("Failed to send pong: %v", err)
return err
}
return nil
})
// Send handshake
modHS := ModHandshake{ServerID: serverID}
modHSData, err := json.Marshal(modHS)
if err != nil {
log.Fatalf("Failed to marshal mod handshake: %v", err)
}
handshake := Handshake{
Type: "mod",
Data: modHSData,
}
if err := conn.WriteJSON(handshake); err != nil {
log.Fatalf("Failed to send handshake: %v", err)
}
log.Println("Handshake sent")
// Read acknowledgment
var ack GatewayAck
if err := conn.ReadJSON(&ack); err != nil {
log.Fatalf("Failed to read acknowledgment: %v", err)
}
log.Printf("Received acknowledgment: status=%s, type=%s", ack.Status, ack.Type)
// Channel for incoming messages
done := make(chan struct{})
// Read loop - handles incoming messages and processes control frames
go func() {
defer close(done)
for {
messageType, message, err := conn.ReadMessage()
if err != nil {
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
log.Printf("WebSocket error: %v", err)
} else {
log.Printf("Connection closed: %v", err)
}
return
}
// Only log text/binary messages (ping/pong handled by handlers)
if messageType == websocket.TextMessage || messageType == websocket.BinaryMessage {
log.Printf("Received from server: %s", string(message))
}
}
}()
// Optional: Send a test message after connecting
time.Sleep(2 * time.Second)
testMsg := GatewayModMessageIn{
MsgID: "test-msg-001",
Server: serverID,
Destination: Destination{
ChannelID: "123456789",
},
Author: MinecraftUser{
ID: "player-uuid-123",
Name: "TestPlayer",
},
Content: "Hello from simulated mod!",
Ts: time.Now().UTC().Format(time.RFC3339),
}
if err := conn.WriteJSON(testMsg); err != nil {
log.Printf("Failed to send test message: %v", err)
} else {
log.Println("Sent test message to gateway")
}
log.Println("Connection established. Responding to pings. Press Ctrl+C to disconnect.")
// Wait for interrupt or connection close
for {
select {
case <-done:
log.Println("Connection closed")
return
case <-interrupt:
log.Println("Interrupt received, closing connection...")
// Send close message
err := conn.WriteMessage(
websocket.CloseMessage,
websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""),
)
if err != nil {
log.Printf("Write close error: %v", err)
return
}
select {
case <-done:
case <-time.After(time.Second):
}
return
}
}
} }

70
util/cache/cache.go vendored Normal file
View File

@@ -0,0 +1,70 @@
package cache
import "sync"
type Cache struct {
mu sync.RWMutex
s2c map[string]string
c2s map[string]string
}
func NewCache() *Cache {
return &Cache{
s2c: make(map[string]string),
c2s: make(map[string]string),
}
}
// Set creates or overwrites the pair a -> b and b -> a.
// It ensures any previous mappings involving a or b are removed first.
func (c *Cache) Set(serverId, channelId string) {
c.mu.Lock()
defer c.mu.Unlock()
if old, ok := c.s2c[serverId]; ok && old != channelId {
delete(c.c2s, old)
}
if old, ok := c.c2s[channelId]; ok && old != serverId {
delete(c.s2c, old)
}
c.s2c[serverId] = channelId
c.c2s[channelId] = serverId
}
func (c *Cache) GetByServerId(serverId string) (string, bool) {
c.mu.RLock()
defer c.mu.RUnlock()
cId, ok := c.s2c[serverId]
return cId, ok
}
func (c *Cache) GetByChannelId(channelId string) (string, bool) {
c.mu.RLock()
defer c.mu.RUnlock()
sId, ok := c.c2s[channelId]
return sId, ok
}
func (c *Cache) RemoveByServerId(serverId string) {
c.mu.RLock()
defer c.mu.RUnlock()
if channelId, ok := c.s2c[serverId]; ok {
delete(c.s2c, serverId)
delete(c.c2s, channelId)
}
}
func (c *Cache) RemoveByChannelId(channelId string) {
c.mu.RLock()
defer c.mu.RUnlock()
if serverId, ok := c.s2c[channelId]; ok {
delete(c.c2s, channelId)
delete(c.s2c, serverId)
}
}

16
util/cache/structs.go vendored Normal file
View File

@@ -0,0 +1,16 @@
package cache
import (
"sync"
)
type ShardedCache struct {
shards []*shard
shardMask uint32
}
type shard struct {
mu sync.RWMutex
s2c map[string]string // serverId -> channelId
c2s map[string]string // channelId -> serverId
}

View File

@@ -1,71 +1,107 @@
// handlers.go
package ws package ws
import ( import (
"context"
"encoding/json" "encoding/json"
"log/slog" "net/http"
"time" "time"
"github.com/gorilla/websocket"
) )
type LoggingModHandler struct { func (wsg *WebsocketGateway) handlePush(w http.ResponseWriter, r *http.Request) {
logger *slog.Logger conn, err := wsg.validateAndUpgradeConnection(w, r)
} if err != nil {
return
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 := ForwardedModMessage{
Type: "mod",
ServerID: msg.Server,
ChannelID: "TODO", // will come from database lookup
User: msg.User,
Content: msg.Content,
Meta: msg.Meta,
Ts: msg.Ts,
ReceivedAt: msg.ReceivedAt,
ForwardedAt: time.Now().UTC(),
} }
b, _ := json.Marshal(fwd) if wsg.bodySizeBytes > 0 {
h.logger.Debug("forwarding mod message", "msg_id", msg.MsgID, "server", msg.Server, "payload", string(b)) conn.SetReadLimit(wsg.bodySizeBytes)
} else {
return nil conn.SetReadLimit(1 << 20) // sensible default 1MiB
}
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 := ForwardedBotMessage{
Type: "bot",
ServerID: "TODO", // will come from database lookup
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) _ = conn.SetReadDeadline(time.Now().Add(60 * time.Second))
h.logger.Debug("forwarding bot message", "msg_id", msg.MsgID, "channel", msg.ChannelID, "payload", string(b)) conn.SetPongHandler(func(appData string) error {
_ = conn.SetReadDeadline(time.Now().Add(60 * time.Second))
return nil 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) {}

View File

@@ -3,6 +3,7 @@ package ws
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"homestead/homestead_gateway/util/cache"
"log/slog" "log/slog"
"sync" "sync"
"time" "time"
@@ -17,25 +18,36 @@ type WebsocketGateway struct {
upgrader websocket.Upgrader upgrader websocket.Upgrader
connsMu sync.Mutex connsMu sync.Mutex
conns map[*websocket.Conn]connMetadata conns map[*websocket.Conn]connectionMetaData
cache cache.Cache
modHandler ModHandler modHandler ModHandler
botHandler BotHandler botHandler BotHandler
logger *slog.Logger logger *slog.Logger
closefn func() error closeFn func() error
} }
type User struct { type MinecraftUser struct {
ID string `json:"id"` ID string `json:"id"`
Name string `json:"name"` Name string `json:"name"`
} }
type DiscordUser struct {
ID string `json:"id"`
Name string `json:"name"`
}
type Destination struct {
ChannelID string `json:"channel_id"`
}
// GatewayModMessageIn : Mod -> Gateway -> Bot // GatewayModMessageIn : Mod -> Gateway -> Bot
type GatewayModMessageIn struct { type GatewayModMessageIn struct {
MsgID string `json:"msg_id"` MsgID string `json:"msg_id"`
Server string `json:"server"` Server string `json:"server"`
User User `json:"user"` Destination Destination `json:"destination"`
Author MinecraftUser `json:"author"`
Content string `json:"content"` Content string `json:"content"`
Meta map[string]interface{} `json:"meta,omitempty"` Meta map[string]interface{} `json:"meta,omitempty"`
Ts string `json:"ts,omitempty"` Ts string `json:"ts,omitempty"`
@@ -46,19 +58,18 @@ type GatewayModMessageIn struct {
type GatewayBotMessageIn struct { type GatewayBotMessageIn struct {
MsgID string `json:"msg_id"` MsgID string `json:"msg_id"`
ChannelID string `json:"channel_id"` ChannelID string `json:"channel_id"`
Author string `json:"author"` Author DiscordUser `json:"author"`
Content string `json:"content"` Content string `json:"content"`
Meta map[string]interface{} `json:"meta,omitempty"` Meta map[string]interface{} `json:"meta,omitempty"`
Ts string `json:"ts,omitempty"` Ts string `json:"ts,omitempty"`
ReceivedAt time.Time `json:"-"` ReceivedAt time.Time `json:"-"` // ReceivedAt is populated by gateway (not from bot)
} }
// ForwardedModMessage : Gateway<GatewayModMessageIn> -> Bot // GatewayModMessageOut : Gateway<GatewayModMessageIn> -> Bot
type ForwardedModMessage struct { type GatewayModMessageOut struct {
Type string `json:"type"` // "mod" Type string `json:"type"` // "mod"
ServerID string `json:"server_id"`
ChannelID string `json:"channel_id"` ChannelID string `json:"channel_id"`
User User `json:"user"` Author MinecraftUser `json:"author"`
Content string `json:"content"` Content string `json:"content"`
Meta map[string]interface{} `json:"meta,omitempty"` Meta map[string]interface{} `json:"meta,omitempty"`
Ts string `json:"ts,omitempty"` Ts string `json:"ts,omitempty"`
@@ -66,12 +77,11 @@ type ForwardedModMessage struct {
ForwardedAt time.Time `json:"forwarded_at"` ForwardedAt time.Time `json:"forwarded_at"`
} }
// ForwardedBotMessage : Gateway<GatewayBotMessageIn> -> Mod // GatewayBotMessageOut : Gateway<GatewayBotMessageIn> -> Mod
type ForwardedBotMessage struct { type GatewayBotMessageOut struct {
Type string `json:"type"` // "bot" Type string `json:"type"` // "bot"
ServerID string `json:"server_id"`
ChannelID string `json:"channel_id"` ChannelID string `json:"channel_id"`
Author string `json:"author"` Author DiscordUser `json:"author"`
Content string `json:"content"` Content string `json:"content"`
Meta map[string]interface{} `json:"meta,omitempty"` Meta map[string]interface{} `json:"meta,omitempty"`
Ts string `json:"ts,omitempty"` Ts string `json:"ts,omitempty"`
@@ -79,25 +89,16 @@ type ForwardedBotMessage struct {
ForwardedAt time.Time `json:"forwarded_at"` ForwardedAt time.Time `json:"forwarded_at"`
} }
type BotAck struct { type GatewayAck struct {
Type string `json:"type"` // "acknowledge" Status string `json:"status"`
MsgID string `json:"msg_id"` Type string `json:"type"`
Status string `json:"status,omitempty"` // "queued"/"sent"
} }
type MessageEnvelope struct { type Handshake struct {
Type string `json:"type"` // "mod" or "bot" Type string `json:"type"` // "mod" or "bot"
Data json.RawMessage `json:"data"` Data json.RawMessage `json:"data"`
} }
type ModHandler interface {
Handle(ctx context.Context, msg GatewayModMessageIn) error
}
type BotHandler interface {
Handle(ctx context.Context, msg GatewayBotMessageIn) error
}
type ModHandshake struct { type ModHandshake struct {
ServerID string `json:"server_id"` ServerID string `json:"server_id"`
} }
@@ -106,7 +107,15 @@ type BotHandshake struct {
BotID string `json:"bot_id"` BotID string `json:"bot_id"`
} }
type connMetadata struct { type ModHandler interface {
connType string // "mod" or "bot" Handle(ctx context.Context, msg GatewayModMessageIn) error
}
type BotHandler interface {
Handle(ctx context.Context, msg GatewayBotMessageIn) error
}
type connectionMetaData struct {
connectionType string // "mod" or "bot"
id string // server_id or bot_id for logging id string // server_id or bot_id for logging
} }

70
ws/temp.go Normal file
View File

@@ -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
}

View File

@@ -2,32 +2,88 @@ package ws
import ( import (
"context" "context"
"fmt" "encoding/json"
"errors"
"log/slog" "log/slog"
"net/http" "net/http"
"os/signal"
"syscall"
"time" "time"
"github.com/gorilla/websocket" "github.com/gorilla/websocket"
) )
func (g *WebsocketGateway) Start() error { // (de-)register
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()
return g.Serve(ctx, fmt.Sprintf(":%d", g.port)) func (wsg *WebsocketGateway) listen(srv *http.Server, addr string, channel chan error) {
wsg.logger.Info("Gateway listening.", "addr", addr)
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
channel <- err
}
close(channel)
}
func (wsg *WebsocketGateway) deafen(srv *http.Server) {
shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
wsg.logger.Info("Shutting down Websocket Gateway.")
_ = srv.Shutdown(shutdownCtx)
wsg.closeAll()
}
// responses
func (wsg *WebsocketGateway) sendHttpError(w http.ResponseWriter, message string, code int) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(code)
_ = json.NewEncoder(w).Encode(map[string]interface{}{"message": message, "code": code})
}
func (wsg *WebsocketGateway) sendWebsocketError(conn *websocket.Conn, message string, code int) {
_ = conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
_ = conn.WriteJSON(map[string]interface{}{"message": message, "code": code})
_ = conn.Close()
}
func (wsg *WebsocketGateway) sendWebsocketResponse(conn *websocket.Conn, content interface{}) error {
_ = conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
if err := conn.WriteJSON(content); err != nil {
wsg.logger.Error("Failed to respond to connection.", "remote", conn.RemoteAddr().String(), "err", err)
_ = conn.Close()
return err
}
return nil
} }
// util // util
func (g *WebsocketGateway) validateApiKey(r *http.Request) bool { func (wsg *WebsocketGateway) validateAndUpgradeConnection(w http.ResponseWriter, r *http.Request) (*websocket.Conn, error) {
if !wsg.validateApiKey(r) {
wsg.sendHttpError(w, "Unauthorized", 401)
wsg.logger.Warn("Authorization failed", "remote", r.RemoteAddr)
return nil, errors.New("unauthorized")
}
conn, err := wsg.upgrader.Upgrade(w, r, nil)
if err != nil {
wsg.logger.Error("Upgrade error.", "remote", r.RemoteAddr)
return nil, err
}
return conn, err
}
func (wsg *WebsocketGateway) validateApiKey(r *http.Request) bool {
apiKey := r.URL.Query().Get("api_key") apiKey := r.URL.Query().Get("api_key")
if apiKey == "" { if apiKey == "" {
apiKey = r.Header.Get("X-API-Key") apiKey = r.Header.Get("X-API-Key")
} }
return !(apiKey == "" || apiKey != g.apiKey) return !(apiKey == "" || apiKey != wsg.apiKey)
} }
func writeJSONSafe(c *websocket.Conn, v interface{}) error { func writeJSONSafe(c *websocket.Conn, v interface{}) error {
@@ -49,24 +105,26 @@ func loggingMiddleware(logger *slog.Logger, next http.Handler) http.Handler {
// connections // connections
func (g *WebsocketGateway) registerConn(c *websocket.Conn, meta connMetadata) { func (wsg *WebsocketGateway) registerConn(c *websocket.Conn, meta connectionMetaData) {
g.connsMu.Lock() wsg.connsMu.Lock()
g.conns[c] = meta wsg.conns[c] = meta
g.connsMu.Unlock() wsg.connsMu.Unlock()
} }
func (g *WebsocketGateway) unregisterConn(c *websocket.Conn) { func (wsg *WebsocketGateway) unregisterConn(c *websocket.Conn) {
g.connsMu.Lock() wsg.connsMu.Lock()
delete(g.conns, c) delete(wsg.conns, c)
g.connsMu.Unlock() wsg.connsMu.Unlock()
} }
func (g *WebsocketGateway) closeAll() { func (wsg *WebsocketGateway) closeAll() {
g.connsMu.Lock() wsg.connsMu.Lock()
defer g.connsMu.Unlock() defer wsg.connsMu.Unlock()
g.logger.Info("closing websocket connections")
for c := range g.conns { wsg.logger.Info("Closing all websocket connections.")
_ = c.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "shutting down"), time.Now().Add(time.Second))
for c := range wsg.conns {
_ = c.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "Shutting down."), time.Now().Add(time.Second))
_ = c.Close() _ = c.Close()
} }
} }

View File

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

View File

@@ -3,11 +3,13 @@ package ws
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"errors" "fmt"
"homestead/homestead_gateway/util/config" "homestead/homestead_gateway/util/config"
"log/slog" "log/slog"
"net" "net"
"net/http" "net/http"
"os/signal"
"syscall"
"time" "time"
"github.com/gorilla/websocket" "github.com/gorilla/websocket"
@@ -16,7 +18,7 @@ import (
func NewWsGateway(cfg config.GatewayConfig, logger *slog.Logger, closefn func() error, modH ModHandler, botH BotHandler) *WebsocketGateway { func NewWsGateway(cfg config.GatewayConfig, logger *slog.Logger, closefn func() error, modH ModHandler, botH BotHandler) *WebsocketGateway {
return &WebsocketGateway{ return &WebsocketGateway{
logger: logger, logger: logger,
closefn: closefn, closeFn: closefn,
apiKey: cfg.Websocket, apiKey: cfg.Websocket,
upgrader: websocket.Upgrader{ upgrader: websocket.Upgrader{
ReadBufferSize: 1024, ReadBufferSize: 1024,
@@ -25,7 +27,7 @@ func NewWsGateway(cfg config.GatewayConfig, logger *slog.Logger, closefn func()
return true // local by default; change for production return true // local by default; change for production
}, },
}, },
conns: make(map[*websocket.Conn]connMetadata), conns: make(map[*websocket.Conn]connectionMetaData),
bodySizeBytes: int64(cfg.BodySize) * 1024 * 1024, bodySizeBytes: int64(cfg.BodySize) * 1024 * 1024,
port: cfg.HttpPort, port: cfg.HttpPort,
modHandler: modH, modHandler: modH,
@@ -33,146 +35,69 @@ func NewWsGateway(cfg config.GatewayConfig, logger *slog.Logger, closefn func()
} }
} }
func (g *WebsocketGateway) Serve(ctx context.Context, listenAddr string) error { func (wsg *WebsocketGateway) Start() error {
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()
return wsg.Serve(ctx, fmt.Sprintf(":%d", wsg.port))
}
func (wsg *WebsocketGateway) Serve(ctx context.Context, listenAddr string) error {
mux := http.NewServeMux() mux := http.NewServeMux()
mux.HandleFunc("/ws", g.handleWS) mux.HandleFunc("/push", wsg.handlePush)
mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) { mux.HandleFunc("/ready", wsg.handleReady)
w.WriteHeader(200) mux.HandleFunc("/health", wsg.handleHealth)
_, _ = w.Write([]byte("ok")) mux.HandleFunc("/register", wsg.handleRegister)
})
srv := &http.Server{ srv := &http.Server{
Addr: listenAddr, Addr: listenAddr,
Handler: loggingMiddleware(g.logger, mux), Handler: loggingMiddleware(wsg.logger, mux),
BaseContext: func(l net.Listener) context.Context { return ctx }, BaseContext: func(l net.Listener) context.Context { return ctx },
} }
errCh := make(chan error, 1) errCh := make(chan error, 1)
go func() {
g.logger.Info("ws gateway listening", "addr", listenAddr) go wsg.listen(srv, listenAddr, errCh)
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
errCh <- err
}
close(errCh)
}()
select { select {
case <-ctx.Done(): case <-ctx.Done():
shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) wsg.deafen(srv)
defer cancel()
g.logger.Info("shutting down http server")
_ = srv.Shutdown(shutdownCtx)
g.closeAll()
return nil return nil
case err := <-errCh: case err := <-errCh:
return err return err
} }
} }
func (g *WebsocketGateway) handleWS(w http.ResponseWriter, r *http.Request) { //
if !g.validateApiKey(r) {
http.Error(w, "unauthorized", http.StatusUnauthorized)
g.logger.Warn("ws auth failed", "remote", r.RemoteAddr)
return
}
conn, err := g.upgrader.Upgrade(w, r, nil) func (wsg *WebsocketGateway) modReadLoop(c *websocket.Conn, meta connectionMetaData) {
if err != nil {
g.logger.Error("ws upgrade error", err, "remote", r.RemoteAddr)
return
}
// Configure read limits & pong handler
if g.bodySizeBytes > 0 {
conn.SetReadLimit(g.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
})
// First message must be a handshake identifying the connection type
typ, data, err := conn.ReadMessage()
if err != nil {
g.logger.Error("failed to read handshake", "err", err, "remote", conn.RemoteAddr().String())
_ = conn.Close()
return
}
if typ != websocket.TextMessage && typ != websocket.BinaryMessage {
g.logger.Warn("invalid handshake message type", "remote", conn.RemoteAddr().String())
_ = writeJSONSafe(conn, map[string]string{"error": "first message must be handshake"})
_ = conn.Close()
return
}
var envelope MessageEnvelope
if err := json.Unmarshal(data, &envelope); err != nil {
g.logger.Warn("invalid handshake json", "err", err, "remote", conn.RemoteAddr().String())
_ = writeJSONSafe(conn, map[string]string{"error": "invalid handshake: " + err.Error()})
_ = conn.Close()
return
}
meta := connMetadata{connType: envelope.Type}
// Validate handshake based on type
switch envelope.Type {
case "mod":
var hs ModHandshake
if err := json.Unmarshal(envelope.Data, &hs); err != nil {
g.logger.Warn("invalid mod handshake", "err", err, "remote", conn.RemoteAddr().String())
_ = writeJSONSafe(conn, map[string]string{"error": "invalid mod handshake: " + err.Error()})
_ = conn.Close()
return
}
meta.id = hs.ServerID
g.registerConn(conn, meta)
g.logger.Info("mod connected", "server_id", hs.ServerID, "remote", conn.RemoteAddr().String())
go g.modReadLoop(conn, meta)
case "bot":
var hs BotHandshake
if err := json.Unmarshal(envelope.Data, &hs); err != nil {
g.logger.Warn("invalid bot handshake", "err", err, "remote", conn.RemoteAddr().String())
_ = writeJSONSafe(conn, map[string]string{"error": "invalid bot handshake: " + err.Error()})
_ = conn.Close()
return
}
meta.id = hs.BotID
g.registerConn(conn, meta)
g.logger.Info("bot connected", "bot_id", hs.BotID, "remote", conn.RemoteAddr().String())
go g.botReadLoop(conn, meta)
default:
g.logger.Warn("unknown connection type", "type", envelope.Type, "remote", conn.RemoteAddr().String())
_ = writeJSONSafe(conn, map[string]string{"error": "unknown connection type: " + envelope.Type})
_ = conn.Close()
return
}
}
func (g *WebsocketGateway) modReadLoop(c *websocket.Conn, meta connMetadata) {
defer func() { defer func() {
g.unregisterConn(c) wsg.unregisterConn(c)
_ = c.Close() _ = c.Close()
g.logger.Info("mod disconnected", "server_id", meta.id, "remote", c.RemoteAddr().String()) wsg.logger.Info("mod disconnected", "server_id", meta.id, "remote", c.RemoteAddr().String())
}() }()
pingTicker := time.NewTicker(30 * time.Second) pingTicker := time.NewTicker(30 * time.Second)
defer pingTicker.Stop() 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 { for {
typ, data, err := c.ReadMessage() typ, data, err := c.ReadMessage()
if err != nil { if err != nil {
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) { if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
g.logger.Warn("unexpected mod close", "server_id", meta.id, "err", err) wsg.logger.Warn("unexpected mod close", "server_id", meta.id, "err", err)
} else { } else {
g.logger.Debug("mod read error", "server_id", meta.id, "err", err) wsg.logger.Debug("mod read error", "server_id", meta.id, "err", err)
} }
return return
} }
@@ -184,56 +109,57 @@ func (g *WebsocketGateway) modReadLoop(c *websocket.Conn, meta connMetadata) {
var msg GatewayModMessageIn var msg GatewayModMessageIn
if err := json.Unmarshal(data, &msg); err != nil { if err := json.Unmarshal(data, &msg); err != nil {
_ = writeJSONSafe(c, map[string]string{"error": "invalid json: " + err.Error()}) _ = writeJSONSafe(c, map[string]string{"error": "invalid json: " + err.Error()})
g.logger.Warn("invalid json from mod", "server_id", meta.id, "remote", c.RemoteAddr().String(), "err", err) wsg.logger.Warn("invalid json from mod", "server_id", meta.id, "remote", c.RemoteAddr().String(), "err", err)
continue continue
} }
msg.ReceivedAt = time.Now().UTC() msg.ReceivedAt = time.Now().UTC()
if err := msg.Validate(); err != nil { if err := msg.Validate(); err != nil {
_ = writeJSONSafe(c, map[string]string{"error": err.Error()}) _ = writeJSONSafe(c, map[string]string{"error": err.Error()})
g.logger.Warn("mod message validation failed", "server_id", meta.id, "remote", c.RemoteAddr().String(), "err", err) wsg.logger.Warn("mod message validation failed", "server_id", meta.id, "remote", c.RemoteAddr().String(), "err", err)
continue continue
} }
// Handle the message (forward to bot, enrich, etc.) // Handle the message (forward to bot, enrich, etc.)
if err := g.modHandler.Handle(context.Background(), msg); err != nil { if err := wsg.modHandler.Handle(context.Background(), msg); err != nil {
_ = writeJSONSafe(c, map[string]string{"error": "handler error: " + err.Error()}) _ = writeJSONSafe(c, map[string]string{"error": "handler error: " + err.Error()})
g.logger.Error("mod handler error", "server_id", meta.id, "err", err) wsg.logger.Error("mod handler error", "server_id", meta.id, "err", err)
continue continue
} }
_ = writeJSONSafe(c, map[string]string{"status": "ok"}) _ = 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 {
g.logger.Debug("write ping failed", "server_id", meta.id, "err", err)
return
}
default:
}
} }
} }
func (g *WebsocketGateway) botReadLoop(c *websocket.Conn, meta connMetadata) { func (wsg *WebsocketGateway) botReadLoop(c *websocket.Conn, meta connectionMetaData) {
defer func() { defer func() {
g.unregisterConn(c) wsg.unregisterConn(c)
_ = c.Close() _ = c.Close()
g.logger.Info("bot disconnected", "bot_id", meta.id, "remote", c.RemoteAddr().String()) wsg.logger.Info("bot disconnected", "bot_id", meta.id, "remote", c.RemoteAddr().String())
}() }()
pingTicker := time.NewTicker(30 * time.Second) pingTicker := time.NewTicker(30 * time.Second)
defer pingTicker.Stop() 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 { for {
typ, data, err := c.ReadMessage() typ, data, err := c.ReadMessage()
if err != nil { if err != nil {
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) { if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
g.logger.Warn("unexpected bot close", "bot_id", meta.id, "err", err) wsg.logger.Warn("unexpected bot close", "bot_id", meta.id, "err", err)
} else { } else {
g.logger.Debug("bot read error", "bot_id", meta.id, "err", err) wsg.logger.Debug("bot read error", "bot_id", meta.id, "err", err)
} }
return return
} }
@@ -245,35 +171,24 @@ func (g *WebsocketGateway) botReadLoop(c *websocket.Conn, meta connMetadata) {
var msg GatewayBotMessageIn var msg GatewayBotMessageIn
if err := json.Unmarshal(data, &msg); err != nil { if err := json.Unmarshal(data, &msg); err != nil {
_ = writeJSONSafe(c, map[string]string{"error": "invalid json: " + err.Error()}) _ = writeJSONSafe(c, map[string]string{"error": "invalid json: " + err.Error()})
g.logger.Warn("invalid json from bot", "bot_id", meta.id, "remote", c.RemoteAddr().String(), "err", err) wsg.logger.Warn("invalid json from bot", "bot_id", meta.id, "remote", c.RemoteAddr().String(), "err", err)
continue continue
} }
msg.ReceivedAt = time.Now().UTC() msg.ReceivedAt = time.Now().UTC()
if err := msg.Validate(); err != nil { if err := msg.Validate(); err != nil {
_ = writeJSONSafe(c, map[string]string{"error": err.Error()}) _ = writeJSONSafe(c, map[string]string{"error": err.Error()})
g.logger.Warn("bot message validation failed", "bot_id", meta.id, "remote", c.RemoteAddr().String(), "err", err) wsg.logger.Warn("bot message validation failed", "bot_id", meta.id, "remote", c.RemoteAddr().String(), "err", err)
continue continue
} }
// Handle the message (forward to mod, enrich, etc.) // Handle the message (forward to mod, enrich, etc.)
if err := g.botHandler.Handle(context.Background(), msg); err != nil { if err := wsg.botHandler.Handle(context.Background(), msg); err != nil {
_ = writeJSONSafe(c, map[string]string{"error": "handler error: " + err.Error()}) _ = writeJSONSafe(c, map[string]string{"error": "handler error: " + err.Error()})
g.logger.Error("bot handler error", "bot_id", meta.id, "err", err) wsg.logger.Error("bot handler error", "bot_id", meta.id, "err", err)
continue continue
} }
_ = writeJSONSafe(c, map[string]string{"status": "ok"}) _ = 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 {
g.logger.Debug("write ping failed", "bot_id", meta.id, "err", err)
return
}
default:
}
} }
} }