continuous send for simulators

bot.go

@@ -5,10 +5,14 @@ package main
 import (
 	"encoding/json"
 	"flag"
+	"fmt"
 	"log"
+	"math/rand"
 	"net/url"
 	"os"
 	"os/signal"
+	"sync"
+	"sync/atomic"
 	"syscall"
 	"time"
 
@@ -18,17 +22,21 @@ import (
 const (
 	gatewayURL = "ws://localhost:3333/sync"
 	apiKey     = "gateway"
-	// must match the mod's channel id used by your mod simulator
 	channelID  = "123456789"
 )
 
+var (
+	minInterval = 500 * time.Millisecond
+	maxInterval = 5 * time.Second
+)
+
 type Handshake struct {
 	Type string          `json:"type"`
 	Data json.RawMessage `json:"data"`
 }
 
 type BotHandshake struct {
-	ChannelId string `json:"channel_id"` // match gateway field exactly
+	ChannelId string `json:"channel_id"`
 }
 
 type GatewayAck struct {
@@ -56,12 +64,21 @@ type GatewayMessageIn struct {
 	ReceivedAt  time.Time              `json:"-"`
 }
 
+func randomDuration(min, max time.Duration) time.Duration {
+	if max <= min {
+		return min
+	}
+	diff := int64(max - min)
+	n := rand.Int63n(diff)
+	return min + time.Duration(n)
+}
+
 func main() {
-	var (
-		botID     = flag.String("bot", "sim-bot-1", "bot id")
-		sendAfter = flag.Duration("send-after", 0, "optional: send a bot->mod message after this delay (e.g. 2s)")
-		sendMsg   = flag.String("msg", "Hello from bot!", "optional bot->mod test message content")
-	)
+	rand.Seed(time.Now().UnixNano())
+
+	botID := flag.String("bot", "sim-bot-1", "bot id")
+	sendAfter := flag.Duration("send-after", 0, "optional: send a bot->mod message after this delay (e.g. 2s)")
+	sendMsg := flag.String("msg", "Hello from bot!", "optional bot->mod test message content")
 	flag.Parse()
 
 	interrupt := make(chan os.Signal, 1)
@@ -75,21 +92,25 @@ func main() {
 	q.Set("api_key", apiKey)
 	u.RawQuery = q.Encode()
 
-	log.Printf("Connecting to %s", u.String())
 	conn, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
 	if err != nil {
 		log.Fatalf("Failed to connect: %v", err)
 	}
-	// we intentionally don't defer conn.Close() here immediately; we'll do on shutdown
-	log.Println("Connected to gateway")
+	defer conn.Close()
+
+	var writeMu sync.Mutex
 
-	// handle server pings by replying a Pong (safe)
 	conn.SetPingHandler(func(appData string) error {
-		log.Println("Received ping from server, sending pong")
-		return conn.WriteControl(websocket.PongMessage, []byte(appData), time.Now().Add(5*time.Second))
+		writeMu.Lock()
+		err := conn.WriteControl(websocket.PongMessage, []byte(appData), time.Now().Add(5*time.Second))
+		writeMu.Unlock()
+		if err != nil {
+			log.Printf("Failed to send pong: %v", err)
+			return err
+		}
+		return nil
 	})
 
-	// send bot handshake (must at least include bot_id)
 	bhs := BotHandshake{ChannelId: channelID}
 	data, err := json.Marshal(bhs)
 	if err != nil {
@@ -97,21 +118,22 @@ func main() {
 		log.Fatalf("Failed to marshal bot handshake: %v", err)
 	}
 	hs := Handshake{Type: "bot", Data: data}
+
+	writeMu.Lock()
 	if err := conn.WriteJSON(hs); err != nil {
+		writeMu.Unlock()
 		_ = conn.Close()
 		log.Fatalf("Failed to send handshake: %v", err)
 	}
-	log.Println("Handshake sent (bot)")
+	writeMu.Unlock()
 
-	// Instead of ReadJSON, read raw message so we can see whatever the server returns (or why it closes).
-	// This avoids a silent parsing failure if server returns non-JSON or closes immediately.
 	conn.SetReadDeadline(time.Now().Add(10 * time.Second))
 	msgType, raw, err := conn.ReadMessage()
 	if err != nil {
 		_ = conn.Close()
 		log.Fatalf("Failed to read handshake response: %v", err)
 	}
-	_ = conn.SetReadDeadline(time.Time{}) // clear deadline
+	_ = conn.SetReadDeadline(time.Time{})
 
 	if msgType == websocket.TextMessage || msgType == websocket.BinaryMessage {
 		log.Printf("Raw handshake reply: %s", string(raw))
@@ -122,11 +144,11 @@ func main() {
 			log.Printf("Parsed ack: status=%q, type=%q", ack.Status, ack.Type)
 		}
 	} else {
-		log.Printf("Handshake reply was control frame or unexpected type=%d", msgType)
+		log.Printf("Handshake reply type=%d", msgType)
 	}
 
-	// From here, start the normal read loop.
 	done := make(chan struct{})
+
 	go func() {
 		defer close(done)
 		for {
@@ -139,20 +161,20 @@ func main() {
 				}
 				return
 			}
-
 			if msgType == websocket.TextMessage || msgType == websocket.BinaryMessage {
 				log.Printf("Received from gateway: %s", string(message))
 			}
 		}
 	}()
 
-	// optionally send a bot->mod message after a delay
+	var msgCounter uint64 = 1
+
 	if *sendAfter > 0 {
 		go func() {
 			time.Sleep(*sendAfter)
 			msg := GatewayMessageIn{
-				MsgID: "bot-msg-001",
-				ID:    channelID, // bot reports channel id as ID
+				MsgID: fmt.Sprintf("bot-msg-%06d", atomic.AddUint64(&msgCounter, 1)),
+				ID:    channelID,
 				Author: User{
 					ID:   *botID,
 					Name: "SimBot",
@@ -160,20 +182,55 @@ func main() {
 				Content: *sendMsg,
 				Ts:      time.Now().UTC(),
 			}
-			// set a write deadline
+			writeMu.Lock()
 			_ = conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
 			if err := conn.WriteJSON(msg); err != nil {
 				log.Printf("Failed to send bot->mod test message: %v", err)
-				return
-			}
+			} else {
 				log.Printf("Sent bot->mod test message (channel=%s)", channelID)
+			}
 			_ = conn.SetWriteDeadline(time.Time{})
+			writeMu.Unlock()
 		}()
 	}
 
-	log.Println("Bot simulator running. Press Ctrl+C to exit.")
+	go func() {
+		for {
+			select {
+			case <-done:
+				return
+			default:
+			}
+			d := randomDuration(minInterval, maxInterval)
+			select {
+			case <-done:
+				return
+			case <-time.After(d):
+				msgNum := atomic.AddUint64(&msgCounter, 1)
+				msg := GatewayMessageIn{
+					MsgID: fmt.Sprintf("sim-bot-msg-%06d", msgNum),
+					ID:    channelID,
+					Author: User{
+						ID:   fmt.Sprintf("%s-%d", *botID, msgNum%1000),
+						Name: fmt.Sprintf("SimBot%d", msgNum%1000),
+					},
+					Content: fmt.Sprintf("Automated bot message #%d (delay %s)", msgNum, d),
+					Ts:      time.Now().UTC(),
+				}
+				writeMu.Lock()
+				_ = conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
+				if err := conn.WriteJSON(msg); err != nil {
+					writeMu.Unlock()
+					log.Printf("Failed to send automated bot message: %v", err)
+					return
+				}
+				_ = conn.SetWriteDeadline(time.Time{})
+				writeMu.Unlock()
+				log.Printf("Sent automated bot message %s", msg.MsgID)
+			}
+		}
+	}()
 
-	// Wait for interrupt or read loop done
 	for {
 		select {
 		case <-done:
@@ -182,8 +239,9 @@ func main() {
 			return
 		case <-interrupt:
 			log.Println("Interrupt received, closing connection...")
-			// politely close
+			writeMu.Lock()
 			_ = conn.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
+			writeMu.Unlock()
 			select {
 			case <-done:
 			case <-time.After(time.Second):

config.toml

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

sim.go

@@ -4,10 +4,14 @@ package main
 
 import (
 	"encoding/json"
+	"fmt"
 	"log"
+	"math/rand"
 	"net/url"
 	"os"
 	"os/signal"
+	"sync"
+	"sync/atomic"
 	"syscall"
 	"time"
 
@@ -22,6 +26,12 @@ const (
 	channelID = "123456789"
 )
 
+// send interval range (random between minInterval and maxInterval)
+var (
+	minInterval = 500 * time.Millisecond
+	maxInterval = 5 * time.Second
+)
+
 type Handshake struct {
 	Type string          `json:"type"`
 	Data json.RawMessage `json:"data"`
@@ -59,6 +69,8 @@ type GatewayMessageIn struct {
 }
 
 func main() {
+	rand.Seed(time.Now().UnixNano())
+
 	interrupt := make(chan os.Signal, 1)
 	signal.Notify(interrupt, os.Interrupt, syscall.SIGTERM)
 
@@ -80,19 +92,19 @@ func main() {
 
 	log.Println("Connected to gateway")
 
-	// respond to pings (server ping -> client must pong). Using SetPingHandler is fine,
-	// but WriteControl for Pong is acceptable too.
+	var writeMu sync.Mutex
+
 	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 {
+		writeMu.Lock()
+		defer writeMu.Unlock()
+		if err := conn.WriteControl(websocket.PongMessage, []byte(appData), time.Now().Add(5*time.Second)); err != nil {
 			log.Printf("Failed to send pong: %v", err)
 			return err
 		}
 		return nil
 	})
 
-	// Build and send handshake including channel id
 	modHS := ModHandshake{
 		ServerID:  serverID,
 		ChannelID: channelID,
@@ -107,12 +119,14 @@ func main() {
 		Data: modHSData,
 	}
 
+	writeMu.Lock()
 	if err := conn.WriteJSON(handshake); err != nil {
+		writeMu.Unlock()
 		log.Fatalf("Failed to send handshake: %v", err)
 	}
+	writeMu.Unlock()
 	log.Println("Handshake sent")
 
-	// Read acknowledgment (some servers might not reply with JSON; handle errors)
 	var ack GatewayAck
 	if err := conn.ReadJSON(&ack); err != nil {
 		log.Fatalf("Failed to read acknowledgment: %v", err)
@@ -129,20 +143,23 @@ func main() {
 				if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
 					log.Printf("WebSocket error: %v", err)
 				} else {
-					log.Printf("Connection closed: %v", err)
+					log.Printf("Connection closed/read error: %v", err)
 				}
 				return
 			}
-			if messageType == websocket.TextMessage || messageType == websocket.BinaryMessage {
+			switch messageType {
+			case websocket.TextMessage, websocket.BinaryMessage:
 				log.Printf("Received from server: %s", string(message))
+			default:
 			}
 		}
 	}()
 
-	// Optional: send a test message after connecting
-	time.Sleep(1 * time.Second)
+	var msgCounter uint64 = 1
+
+	func() {
 		testMsg := GatewayMessageIn{
-		MsgID: "test-msg-001",
+			MsgID: fmt.Sprintf("test-msg-%06d", atomic.AddUint64(&msgCounter, 1)),
 			ID:    serverID,
 			Destination: Destination{
 				ID: channelID,
@@ -155,36 +172,86 @@ func main() {
 			Ts:      time.Now().UTC(),
 		}
 
+		writeMu.Lock()
 		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("Sent initial test message to gateway")
+		}
+		writeMu.Unlock()
+	}()
+
+	go func() {
+		for {
+			d := randomDuration(minInterval, maxInterval)
+
+			select {
+			case <-done:
+				return
+			case <-time.After(d):
+				// build message
+				msgNum := atomic.AddUint64(&msgCounter, 1)
+				msg := GatewayMessageIn{
+					MsgID: fmt.Sprintf("sim-msg-%06d", msgNum),
+					ID:    serverID,
+					Destination: Destination{
+						ID: channelID,
+					},
+					Author: User{
+						ID:   fmt.Sprintf("sim-user-%d", msgNum%1000),
+						Name: fmt.Sprintf("SimUser%d", msgNum%1000),
+					},
+					Content: fmt.Sprintf("Random interval message #%d (delay %s)", msgNum, d),
+					Ts:      time.Now().UTC(),
 				}
 
-	log.Println("Connection established. Responding to pings. Press Ctrl+C to disconnect.")
+				writeMu.Lock()
+				if err := conn.WriteJSON(msg); err != nil {
+					log.Printf("Failed to send simulated message: %v", err)
+					writeMu.Unlock()
+					return
+				}
+				writeMu.Unlock()
+
+				log.Printf("Sent simulated message %s (next wait up to %s)", msg.MsgID, maxInterval)
+			}
+		}
+	}()
+
+	log.Println("Connection established. Sending simulated messages at random intervals. Press Ctrl+C to disconnect.")
 
 	for {
 		select {
 		case <-done:
-			log.Println("Connection closed")
+			log.Println("Connection read loop closed, exiting")
 			return
 		case <-interrupt:
 			log.Println("Interrupt received, closing connection...")
 
+			writeMu.Lock()
 			err := conn.WriteMessage(
 				websocket.CloseMessage,
 				websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""),
 			)
+			writeMu.Unlock()
 			if err != nil {
 				log.Printf("Write close error: %v", err)
-				return
 			}
 
 			select {
 			case <-done:
-			case <-time.After(time.Second):
+			case <-time.After(1 * time.Second):
 			}
 			return
 		}
 	}
 }
+
+func randomDuration(min, max time.Duration) time.Duration {
+	if max <= min {
+		return min
+	}
+	diff := int64(max - min)
+	n := rand.Int63n(diff)
+	return min + time.Duration(n)
+}

util/config/structs.go

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

ws/registry.go

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

ws/structs.go

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

ws/util.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"log/slog"
 	"net/http"
 	"strings"
 	"time"
@@ -99,16 +98,26 @@ func (wsg *WebsocketGateway) validateApiKey(r *http.Request) bool {
 	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) {
 		start := time.Now()
 		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
 
+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 {
 	if typ == "bot" {
 		wsg.registry.botMu.Lock()
@@ -140,20 +149,23 @@ func (wsg *WebsocketGateway) closeAll() {
 
 	wsg.registry.UnregisterBot()
 
-	wsg.registry.mu.RLock()
-	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.ForEach(func(channelID string) {
 		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 {
 	return &Registry{
 		entries:  make(map[string]*ChannelEntry),

ws/websocket.go

@@ -21,14 +21,8 @@ func NewWebsocketGateway(cfg config.GatewayConfig, logger *slog.Logger, closefn
 		closeFn:       closefn,
 		port:          cfg.HttpPort,
 		apiKey:        cfg.Websocket,
-		registry: NewRegistry(32),
-		upgrader: websocket.Upgrader{
-			ReadBufferSize:  1024,
-			WriteBufferSize: 1024,
-			CheckOrigin: func(r *http.Request) bool {
-				return true // local by default; change for production
-			},
-		},
+		upgrader:      NewUpgrader(),
+		registry:      NewRegistry(cfg.QueueSize),
 		bodySizeBytes: int64(cfg.BodySize) * 1024 * 1024,
 	}
 }
@@ -47,7 +41,7 @@ func (wsg *WebsocketGateway) Serve(ctx context.Context, listenAddr string) error
 
 	srv := &http.Server{
 		Addr:        listenAddr,
-		Handler:     loggingMiddleware(wsg.logger, mux),
+		Handler:     wsg.loggingMiddleware(mux),
 		BaseContext: func(l net.Listener) context.Context { return ctx },
 	}
 	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) {
 	defer func() {
 		wsg.unregisterConn(_type, channelId)