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

quality of life, updated config structure

Browse Source
This commit is contained in:
Overlord
2025-12-08 06:57:54 +01:00
parent 8f75e6491f
commit 786b217f05
6 changed files with 77 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
config.toml
View File

@@ -7,10 +7,4 @@ rotation = 3 # in days
http_port = 3333 http_port = 3333
websocket = "gateway" websocket = "gateway"
body_size = 2 # in MB body_size = 2 # in MB
queue_max = 8192 queue_max = 32
[database]
host_dsn = ""
username = ""
password = ""
database = ""

12
util/config/structs.go
View File

@@ -3,9 +3,8 @@ package config
import "log/slog" import "log/slog"
type Config struct { type Config struct {
Log LogConfig `toml:"log"` Log LogConfig `toml:"log"`
Gateway GatewayConfig `toml:"gateway"` Gateway GatewayConfig `toml:"gateway"`
Database DatabaseConfig `toml:"database"`
} }
type GatewayConfig struct { type GatewayConfig struct {
@@ -20,10 +19,3 @@ type LogConfig struct {
Directory string `toml:"directory"` Directory string `toml:"directory"`
Rotation int `toml:"rotation"` Rotation int `toml:"rotation"`
} }
type DatabaseConfig struct {
HostDSN string `toml:"host_dsn"`
Username string `toml:"username"`
Password string `toml:"password"`
Database string `toml:"database"`
}

47
ws/registry.go
View File

@@ -13,14 +13,17 @@ func (q *BoundedQueue) Enqueue(m GatewayMessageOut) bool {
if q.capacity == 0 { if q.capacity == 0 {
return false return false
} }
if q.length < q.capacity { if q.length < q.capacity {
q.buf[(q.start+q.length)%q.capacity] = m q.buf[(q.start+q.length)%q.capacity] = m
q.length++ q.length++
return true return true
} }
// overwrite oldest // overwrite oldest
q.buf[q.start] = m q.buf[q.start] = m
q.start = (q.start + 1) % q.capacity q.start = (q.start + 1) % q.capacity
return true return true
} }
@@ -30,12 +33,15 @@ func (q *BoundedQueue) PopAll() []GatewayMessageOut {
if q.length == 0 { if q.length == 0 {
return nil return nil
} }
out := make([]GatewayMessageOut, 0, q.length) out := make([]GatewayMessageOut, 0, q.length)
for i := 0; i < q.length; i++ { for i := 0; i < q.length; i++ {
out = append(out, q.buf[(q.start+i)%q.capacity]) out = append(out, q.buf[(q.start+i)%q.capacity])
} }
q.start = 0 q.start = 0
q.length = 0 q.length = 0
return out return out
} }
@@ -64,6 +70,19 @@ func (r *Registry) getOrCreate(channel string) *ChannelEntry {
return e return e
} }
func (r *Registry) ForEach(cb func(channelID string)) {
r.mu.RLock()
ids := make([]string, 0, len(r.entries))
for id := range r.entries {
ids = append(ids, id)
}
r.mu.RUnlock()
for _, id := range ids {
cb(id)
}
}
// //
// RegisterMod : map channel_id -> mod conn (serverID) // RegisterMod : map channel_id -> mod conn (serverID)
@@ -71,9 +90,12 @@ func (r *Registry) RegisterMod(channelID, serverID string, conn *websocket.Conn)
e := r.getOrCreate(channelID) e := r.getOrCreate(channelID)
e.mu.Lock() e.mu.Lock()
defer e.mu.Unlock() defer e.mu.Unlock()
if e.Mod != nil && e.Mod.Conn != nil { if e.Mod != nil && e.Mod.Conn != nil {
_ = e.Mod.Conn.Close() _ = e.Mod.Conn.Close()
} }
e.Mod = &ConnWrapper{Conn: conn, ServerID: serverID, LastSeen: time.Now()} e.Mod = &ConnWrapper{Conn: conn, ServerID: serverID, LastSeen: time.Now()}
// flush queued bot->mod messages for this channel // flush queued bot->mod messages for this channel
// caller should use FlushChannelWithSender to perform actual sends // caller should use FlushChannelWithSender to perform actual sends
@@ -82,9 +104,11 @@ func (r *Registry) RegisterMod(channelID, serverID string, conn *websocket.Conn)
// RegisterBot : single connection for bot. after registration call FlushAllToBotWithSender // RegisterBot : single connection for bot. after registration call FlushAllToBotWithSender
func (r *Registry) RegisterBot(conn *websocket.Conn) { func (r *Registry) RegisterBot(conn *websocket.Conn) {
r.botMu.Lock() r.botMu.Lock()
if r.bot != nil && r.bot.Conn != nil { if r.bot != nil && r.bot.Conn != nil {
_ = r.bot.Conn.Close() _ = r.bot.Conn.Close()
} }
r.bot = &ConnWrapper{Conn: conn, LastSeen: time.Now()} r.bot = &ConnWrapper{Conn: conn, LastSeen: time.Now()}
r.botMu.Unlock() r.botMu.Unlock()
} }
@@ -103,13 +127,7 @@ func (r *Registry) UnregisterMod(channelID string) {
e.mu.Unlock() e.mu.Unlock()
if modConn != nil && modConn.Conn != nil { if modConn != nil && modConn.Conn != nil {
_ = modConn.Conn.SetWriteDeadline(time.Now().Add(time.Second)) closeConn(modConn.Conn)
_ = modConn.Conn.WriteControl(
websocket.CloseMessage,
websocket.FormatCloseMessage(websocket.CloseNormalClosure, "Disconnecting."),
time.Now().Add(time.Second),
)
_ = modConn.Conn.Close()
} }
} }
@@ -120,13 +138,7 @@ func (r *Registry) UnregisterBot() {
r.botMu.Unlock() r.botMu.Unlock()
if botConn != nil && botConn.Conn != nil { if botConn != nil && botConn.Conn != nil {
_ = botConn.Conn.SetWriteDeadline(time.Now().Add(time.Second)) closeConn(botConn.Conn)
_ = botConn.Conn.WriteControl(
websocket.CloseMessage,
websocket.FormatCloseMessage(websocket.CloseNormalClosure, "Disconnecting."),
time.Now().Add(time.Second),
)
_ = botConn.Conn.Close()
} }
} }
@@ -135,6 +147,7 @@ func (r *Registry) Send(channelID string, out GatewayMessageOut, sendOverConn fu
r.botMu.Lock() r.botMu.Lock()
b := r.bot b := r.bot
r.botMu.Unlock() r.botMu.Unlock()
if b != nil && b.Conn != nil { if b != nil && b.Conn != nil {
if err := sendOverConn(b.Conn, out); err == nil { if err := sendOverConn(b.Conn, out); err == nil {
return true, false, nil return true, false, nil
@@ -147,6 +160,7 @@ func (r *Registry) Send(channelID string, out GatewayMessageOut, sendOverConn fu
e.mu.Lock() e.mu.Lock()
enq := e.Queue.Enqueue(out) enq := e.Queue.Enqueue(out)
e.mu.Unlock() e.mu.Unlock()
if !enq { if !enq {
return false, false, fmt.Errorf("queue disabled") return false, false, fmt.Errorf("queue disabled")
} }
@@ -157,6 +171,7 @@ func (r *Registry) Send(channelID string, out GatewayMessageOut, sendOverConn fu
e.mu.Lock() e.mu.Lock()
mod := e.Mod mod := e.Mod
e.mu.Unlock() e.mu.Unlock()
if mod != nil && mod.Conn != nil { if mod != nil && mod.Conn != nil {
if err := sendOverConn(mod.Conn, out); err == nil { if err := sendOverConn(mod.Conn, out); err == nil {
return true, false, nil return true, false, nil
@@ -168,6 +183,7 @@ func (r *Registry) Send(channelID string, out GatewayMessageOut, sendOverConn fu
e.mu.Lock() e.mu.Lock()
enq := e.Queue.Enqueue(out) enq := e.Queue.Enqueue(out)
e.mu.Unlock() e.mu.Unlock()
if !enq { if !enq {
return false, false, fmt.Errorf("queue disabled") return false, false, fmt.Errorf("queue disabled")
} }
@@ -183,11 +199,13 @@ func (r *Registry) FlushChannelWithSender(channelID string, sendOverConn func(*w
if e == nil { if e == nil {
return return
} }
e.mu.Lock() e.mu.Lock()
if e.Mod == nil || e.Mod.Conn == nil { if e.Mod == nil || e.Mod.Conn == nil {
e.mu.Unlock() e.mu.Unlock()
return return
} }
msgs := e.Queue.PopAll() msgs := e.Queue.PopAll()
modConn := e.Mod.Conn modConn := e.Mod.Conn
e.mu.Unlock() e.mu.Unlock()
@@ -221,6 +239,7 @@ func (r *Registry) FlushAllToBotWithSender(sendOverConn func(*websocket.Conn, Ga
e.mu.Lock() e.mu.Lock()
msgs := e.Queue.PopAll() msgs := e.Queue.PopAll()
e.mu.Unlock() e.mu.Unlock()
if len(msgs) == 0 { if len(msgs) == 0 {
continue continue
} }

20
ws/structs.go
View File

@@ -67,19 +67,19 @@ type Destination struct {
type GatewayMessageIn struct { type GatewayMessageIn struct {
Type string Type string
ID string `json:"id"` // where am I from (channel_id or server_id) ID string `json:"id"`
MsgID string `json:"msg_id"` // msg id MsgID string `json:"msg_id"`
Destination Destination `json:"destination,omitempty"` // where do I wanna go (channel_id or empty if from Bot) Destination Destination `json:"destination,omitempty"`
Author User `json:"author"` // who sent the message Author User `json:"author"`
Content string `json:"content"` // message content Content string `json:"content"`
Meta map[string]interface{} `json:"meta,omitempty"` // additional metadata Meta map[string]interface{} `json:"meta,omitempty"`
Ts time.Time `json:"ts,omitempty"` // timestamp Ts time.Time `json:"ts,omitempty"`
ReceivedAt time.Time `json:"-"` // ReceivedAt is populated by gateway (not from mod) ReceivedAt time.Time `json:"-"`
} }
type GatewayMessageOut struct { type GatewayMessageOut struct {
Type string `json:"type"` // "mod"|"bot" Type string `json:"type"`
ID string `json:"channel_id,omitempty"` // message.Destination.ID ID string `json:"channel_id,omitempty"`
Author User `json:"author"` Author User `json:"author"`
Content string `json:"content"` Content string `json:"content"`
Meta map[string]interface{} `json:"meta,omitempty"` Meta map[string]interface{} `json:"meta,omitempty"`

36
ws/util.go
View File

@@ -4,7 +4,6 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"errors" "errors"
"log/slog"
"net/http" "net/http"
"strings" "strings"
"time" "time"
@@ -99,16 +98,26 @@ func (wsg *WebsocketGateway) validateApiKey(r *http.Request) bool {
return !(apiKey == "" || apiKey != wsg.apiKey) return !(apiKey == "" || apiKey != wsg.apiKey)
} }
func loggingMiddleware(logger *slog.Logger, next http.Handler) http.Handler { func (wsg *WebsocketGateway) loggingMiddleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
start := time.Now() start := time.Now()
next.ServeHTTP(w, r) next.ServeHTTP(w, r)
logger.Info("Incoming HTTP request.", "remote", r.RemoteAddr, "path", r.URL.Path, "duration", time.Since(start)) wsg.logger.Info("Incoming HTTP request.", "remote", r.RemoteAddr, "path", r.URL.Path, "duration", time.Since(start))
}) })
} }
// connections // connections
func closeConn(conn *websocket.Conn) {
_ = conn.SetWriteDeadline(time.Now().Add(time.Second))
_ = conn.WriteControl(
websocket.CloseMessage,
websocket.FormatCloseMessage(websocket.CloseNormalClosure, "Disconnecting."),
time.Now().Add(time.Second),
)
_ = conn.Close()
}
func (wsg *WebsocketGateway) registerConn(conn *websocket.Conn, typ, channelId, serverId string) bool { func (wsg *WebsocketGateway) registerConn(conn *websocket.Conn, typ, channelId, serverId string) bool {
if typ == "bot" { if typ == "bot" {
wsg.registry.botMu.Lock() wsg.registry.botMu.Lock()
@@ -140,20 +149,23 @@ func (wsg *WebsocketGateway) closeAll() {
wsg.registry.UnregisterBot() wsg.registry.UnregisterBot()
wsg.registry.mu.RLock() wsg.registry.ForEach(func(channelID string) {
channelIDs := make([]string, 0, len(wsg.registry.entries))
for channelID := range wsg.registry.entries {
channelIDs = append(channelIDs, channelID)
}
wsg.registry.mu.RUnlock()
for _, channelID := range channelIDs {
wsg.registry.UnregisterMod(channelID) wsg.registry.UnregisterMod(channelID)
} })
} }
// //
func NewUpgrader() websocket.Upgrader {
return websocket.Upgrader{
ReadBufferSize: 1024,
WriteBufferSize: 1024,
CheckOrigin: func(r *http.Request) bool {
return true // local by default; change for production
},
}
}
func NewRegistry(queueCap int) *Registry { func NewRegistry(queueCap int) *Registry {
return &Registry{ return &Registry{
entries: make(map[string]*ChannelEntry), entries: make(map[string]*ChannelEntry),

135
ws/websocket.go
View File

@@ -17,18 +17,12 @@ import (
func NewWebsocketGateway(cfg config.GatewayConfig, logger *slog.Logger, closefn func() error) *WebsocketGateway { func NewWebsocketGateway(cfg config.GatewayConfig, logger *slog.Logger, closefn func() error) *WebsocketGateway {
return &WebsocketGateway{ return &WebsocketGateway{
logger: logger, logger: logger,
closeFn: closefn, closeFn: closefn,
port: cfg.HttpPort, port: cfg.HttpPort,
apiKey: cfg.Websocket, apiKey: cfg.Websocket,
registry: NewRegistry(32), upgrader: NewUpgrader(),
upgrader: websocket.Upgrader{ registry: NewRegistry(cfg.QueueSize),
ReadBufferSize: 1024,
WriteBufferSize: 1024,
CheckOrigin: func(r *http.Request) bool {
return true // local by default; change for production
},
},
bodySizeBytes: int64(cfg.BodySize) * 1024 * 1024, bodySizeBytes: int64(cfg.BodySize) * 1024 * 1024,
} }
} }
@@ -47,7 +41,7 @@ func (wsg *WebsocketGateway) Serve(ctx context.Context, listenAddr string) error
srv := &http.Server{ srv := &http.Server{
Addr: listenAddr, Addr: listenAddr,
Handler: loggingMiddleware(wsg.logger, mux), Handler: wsg.loggingMiddleware(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)
@@ -65,121 +59,6 @@ func (wsg *WebsocketGateway) Serve(ctx context.Context, listenAddr string) error
// //
//func (wsg *WebsocketGateway) modReadLoop(conn *websocket.Conn, meta cache.ConnectionMetaData) {
// defer func() {
// wsg.unregisterConn(conn, meta, "mod")
// wsg.logger.Info("Client disconnected.", "remote", conn.RemoteAddr().String(), "server_id", meta.ID)
// }()
//
// ticker := time.NewTicker(30 * time.Second)
// defer ticker.Stop()
//
// go func() {
// for range ticker.C {
// wsg.sendWebsocketPing(conn)
// }
// }()
//
// for {
// typ, data, err := conn.ReadMessage()
//
// if err != nil {
// if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
// wsg.logger.Warn("Mod-Client unexpectedly closed the connection.", "err", err)
// }
// return
// }
//
// if typ != websocket.TextMessage && typ != websocket.BinaryMessage {
// continue
// }
//
// var msg GatewayModMessageIn
// if err := json.Unmarshal(data, &msg); err != nil {
// _ = 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(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(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(conn, map[string]string{"status": "completed"}) // or "queued"
// }
//}
//
//func (wsg *WebsocketGateway) botReadLoop(conn *websocket.Conn, meta cache.ConnectionMetaData) {
// defer func() {
// wsg.unregisterConn(conn, meta, "bot")
// wsg.logger.Info("bot disconnected", "bot_id", meta.ID, "remote", conn.RemoteAddr().String())
// }()
//
// pingTicker := time.NewTicker(30 * time.Second)
// defer pingTicker.Stop()
//
// // Send pings in a separate goroutine
// go func() {
// for range pingTicker.C {
// _ = 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)
// }
// }()
//
// for {
// 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)
// } else {
// wsg.logger.Debug("bot read error", "bot_id", meta.ID, "err", err)
// }
// return
// }
//
// if typ != websocket.TextMessage && typ != websocket.BinaryMessage {
// continue
// }
//
// var msg GatewayBotMessageIn
// if err := json.Unmarshal(data, &msg); err != nil {
// _ = 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(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(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(conn, map[string]string{"status": "ok"})
// }
//}
func (wsg *WebsocketGateway) read(conn *websocket.Conn, _type, channelId string) { func (wsg *WebsocketGateway) read(conn *websocket.Conn, _type, channelId string) {
defer func() { defer func() {
wsg.unregisterConn(_type, channelId) wsg.unregisterConn(_type, channelId)