Merge tag 'beta-1.1' into develop

Tagging version beta-1.1 beta-1.1

.gitignore

@@ -1,3 +1,7 @@
+*
+!*/
+!*.*
+
 *.exe
 *.exe~
 *.dll

README.md

@@ -2,14 +2,476 @@
 
 Gateway between multiple HomesteadRelay's and the HomesteadToGo Bot.
 
+---
 
-## dev notes
+## HomesteadGateway Developer Documentation
 
-perhaps drop database, instead
+## Overview
+
+HomesteadGateway is a WebSocket-based message routing gateway that facilitates bidirectional communication between game server mods/plugins and external bots (e.g., Discord bots). It acts as a relay, routing messages based on channel identifiers and managing message queues when endpoints are offline.
+
+**Key Features:**
+- WebSocket-based real-time communication
+- Channel-based message routing
+- Automatic message queuing for offline recipients
+- API key authentication
+- Connection keep-alive via ping/pong
+- Support for multiple concurrent mod connections per channel
+
+---
+
+## Quick Start
+
+1. Start the gateway (build/run your application that embeds the WebSocket gateway).
+2. Connect a client to the WebSocket endpoint:
+    - URL: `ws://<host>:<port>/sync?api_key=<your_api_key>`
+3. Immediately send a handshake JSON with type `mod` or `bot`.
+4. On success, you’ll receive `{"status":"connected","type":"mod|bot"}`.
+5. Exchange messages as JSON.
+
+See the full details below.
+
+---
+
+## Architecture
+
+### Connection Types
+
+The gateway supports two types of connections:
+
+1. **Mod Connection** - Game server mods/plugins that send and receive player messages
+2. **Bot Connection** - External bots (typically Discord) that bridge messages to/from other platforms
+
+### Message Flow
 
 ```
-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
+Mod (Server) ←→ Gateway ←→ Bot (Discord)
+     ↓                          ↓
+  Channel A                 Channel A
 ```
+
+Messages are routed based on `channel_id`:
+- **Mod → Bot**: Messages from a mod are forwarded to the bot
+- **Bot → Mod**: Messages from a bot are forwarded to the mod registered for that channel
+
+---
+
+## Connection Setup
+
+### Endpoint
+
+```
+ws://<host>:<port>/sync?api_key=<your_api_key>
+```
+
+**Default Port:** 3333 (configurable)
+
+### Authentication
+
+Authentication is performed via API key, which can be provided in two ways:
+
+1. **Query Parameter** (recommended):
+   ```
+   ws://localhost:3333/sync?api_key=gateway
+   ```
+
+2. **HTTP Header**:
+   ```
+   X-API-Key: gateway
+   ```
+
+### Connection Timeout
+
+After connecting, you **must** send a handshake message within **60 seconds** or the connection will be closed.
+
+---
+
+## Handshake Protocol
+
+### Step 1: Establish WebSocket Connection
+
+Connect to the `/sync` endpoint with your API key.
+
+### Step 2: Send Handshake Message
+
+Immediately after connecting, send a JSON handshake message:
+
+```json
+{
+  "type": "mod",  // or "bot"
+  "data": { ... }
+}
+```
+
+#### Mod Handshake
+
+For game server mods/plugins:
+
+```json
+{
+  "type": "mod",
+  "data": {
+    "server_id": "minecraft-server-001",
+    "channel_id": "123456789"
+  }
+}
+```
+
+**Fields:**
+- `server_id` (string, required): Unique identifier for your server instance
+- `channel_id` (string, required): The Discord channel ID (or equivalent) this mod serves
+
+#### Bot Handshake
+
+For bots (Discord bots, etc.):
+
+```json
+{
+  "type": "bot",
+  "data": {
+    "bot_id": "discord-bot-123"
+  }
+}
+```
+
+**Fields:**
+- `bot_id` (string, required): The bots ID
+
+**Note:** Only **one bot connection** is allowed at a time. Trying to connect a new bot will result in an `409 Conflict` Error.
+
+### Step 3: Receive Acknowledgment
+
+After sending the handshake, wait for an acknowledgment:
+
+**Success Response:**
+```json
+{
+  "status": "connected",
+  "type": "mod|bot"
+}
+```
+
+**Error Responses:**
+```json
+{
+  "message": "Malformed handshake.",
+  "code": 400
+}
+```
+
+```json
+{
+  "message": "Bot already connected.",
+  "code": 409
+}
+```
+
+### Step 4: Begin Message Exchange
+
+Once acknowledged, the connection is established and you can start sending/receiving messages.
+
+---
+
+## Message Protocol
+
+### Sending Messages
+
+After handshake, send messages as JSON:
+
+#### From Mod to Bot
+
+```json
+{
+  "msg_id": "msg-unique-123",
+  "id": "minecraft-server-001",
+  "destination": {
+    "channel_id": "123456789"
+  },
+  "author": {
+    "id": "player-uuid-abc",
+    "name": "PlayerName"
+  },
+  "content": "Hello from the game server!",
+  "meta": {
+    "server_name": "Survival Server",
+    "world": "overworld"
+  },
+  "ts": "2025-12-08T10:30:00Z"
+}
+```
+
+**Required Fields:**
+- `msg_id` (string): Unique message identifier (generate client-side)
+- `id` (string): Server ID (must match your handshake `server_id`)
+- `destination.channel_id` (string): Target channel ID
+- `author.id` (string): User/player identifier
+- `content` (string): Message content (non-empty)
+
+**Optional Fields:**
+- `author.name` (string): Display name for the author
+- `meta` (object): Additional metadata (arbitrary key-value pairs)
+- `ts` (RFC3339 timestamp): Message timestamp (defaults to server time if omitted)
+
+#### From Bot to Mod
+
+```json
+{
+  "msg_id": "discord-msg-456",
+  "id": "123456789",
+  "author": {
+    "id": "discord-user-789",
+    "name": "DiscordUser"
+  },
+  "content": "Hello from Discord!",
+  "meta": {
+    "platform": "discord",
+    "roles": ["admin"]
+  },
+  "ts": "2025-12-08T10:31:00Z"
+}
+```
+
+**Required Fields:**
+- `msg_id` (string): Unique message identifier
+- `id` (string): Channel ID (from which channel the message originates)
+- `author.id` (string): User identifier
+- `content` (string): Message content (non-empty)
+
+**Optional Fields:**
+- `author.name` (string): Display name
+- `meta` (object): Additional metadata
+- `ts` (RFC3339 timestamp): Message timestamp
+
+**Note:** Bot messages do **not** include a `destination` field, as the channel ID in `id` determines routing.
+
+### Receiving Messages
+
+Messages are received as JSON in the same format they were sent:
+
+#### Mod Receives (from Bot)
+
+```json
+{
+  "type": "bot",
+  "channel_id": "123456789",
+  "author": {
+    "id": "discord-user-789",
+    "name": "DiscordUser"
+  },
+  "content": "Hello from Discord!",
+  "meta": {
+    "platform": "discord"
+  },
+  "ts": "2025-12-08T10:31:00Z",
+  "received_at": "2025-12-08T10:31:00.123Z",
+  "forwarded_at": "2025-12-08T10:31:00.125Z"
+}
+```
+
+#### Bot Receives (from Mod)
+
+```json
+{
+  "type": "mod",
+  "channel_id": "123456789",
+  "author": {
+    "id": "player-uuid-abc",
+    "name": "PlayerName"
+  },
+  "content": "Hello from the game server!",
+  "meta": {
+    "server_name": "Survival Server"
+  },
+  "ts": "2025-12-08T10:30:00Z",
+  "received_at": "2025-12-08T10:30:00.100Z",
+  "forwarded_at": "2025-12-08T10:30:00.102Z"
+}
+```
+
+**Additional Fields in Received Messages:**
+- `type` (string): Origin type ("mod" or "bot")
+- `channel_id` (string): The channel this message belongs to
+- `received_at` (RFC3339): When gateway received the message
+- `forwarded_at` (RFC3339): When gateway forwarded the message
+
+### Message Acknowledgments
+
+After sending each message, you'll receive an acknowledgment:
+
+```json
+{
+  "status": "completed",
+  "type": "mod"
+}
+```
+
+**Status Values:**
+
+- `completed`: Message was delivered immediately to recipient
+- `queued`: Recipient is offline; message queued for later delivery
+- `failed`: Message could not be delivered or queued
+
+---
+
+## Message Queuing
+
+When a recipient is offline, messages are automatically queued:
+
+- **Queue Size:** Configurable (default: 8 messages per channel)
+- **Queue Behavior:** Circular buffer (oldest messages are overwritten when full)
+- **Flush Trigger:** When recipient reconnects, all queued messages are delivered
+
+### Example Flow
+
+1. Mod sends message while bot is offline → Message queued
+2. Bot connects and completes handshake → All queued messages flushed to bot
+3. Bot sends message while mod is offline → Message queued
+4. Mod connects → Queued messages flushed to mod
+
+---
+
+## Keep-Alive & Ping/Pong
+
+The gateway sends **WebSocket ping messages every 30 seconds** to maintain connections.
+
+**Client Responsibilities:**
+1. **Respond to pings**: Your WebSocket library should automatically handle pong responses
+2. **Handle pongs**: Set a pong handler to reset read deadlines
+3. **Read Deadline**: The gateway sets a 60-second read deadline, reset on each pong
+
+### Example (Go)
+
+```go
+conn.SetPongHandler(func(appData string) error {
+    conn.SetReadDeadline(time.Now().Add(60 * time.Second))
+    return nil
+})
+```
+
+---
+
+## Error Handling
+
+### Connection Errors
+
+**Handshake Errors:**
+- `400` - Malformed handshake JSON
+- `401` - Invalid or missing API key
+- `409` - Bot already connected (only for bot handshakes)
+- `500` - Internal server error
+
+**Message Errors:**
+- `400` - Malformed message (missing required fields)
+
+Errors are sent as JSON:
+```json
+{
+  "message": "Malformed message.",
+  "code": 400
+}
+```
+
+### Validation Rules
+
+Messages are validated on receipt:
+- `id` must not be empty
+- `msg_id` must not be empty
+- `author.id` must not be empty
+- `content` must not be empty
+- For mod messages: `destination.channel_id` must not be empty
+
+### Websocket Closures
+
+- `1000` - Normal closure
+- `1001` - Going away
+
+Handle other WebSocket closures as unexpected errors.
+
+---
+
+## Rate Limits & Restrictions
+
+- **Message Size Limit:** 2 MB per message (configurable)
+- **Read Limit:** Messages exceeding the limit will close the connection
+- **Concurrent Mods:** Multiple mods *can* connect to the same channel ID (different server IDs)
+- **Concurrent Bots:** Only **one bot connection** allowed globally
+
+---
+
+## Best Practices
+
+### 1. Generate Unique Message IDs
+Always generate unique `msg_id` values for each message. Consider using:
+- UUID v4
+- Timestamp + random suffix
+- Sequential counter with prefix
+
+### 2. Handle Reconnections
+Implement automatic reconnection logic with exponential backoff:
+```
+1st retry: 1 second
+2nd retry: 2 seconds
+3rd retry: 4 seconds
+Max: 30 seconds
+```
+
+### 3. Set Appropriate Timeouts
+- Write timeout: 5 seconds (same as Gateway)
+- Read timeout: 60 seconds (reset on pong)
+
+### 4. Validate Before Sending
+Check required fields locally before sending to avoid validation errors.
+
+### 5. Monitor Acknowledgments
+Track acknowledgment statuses:
+- `completed`: Message delivered
+- `queued`: Message queued
+- `failed`: Log and potentially retry
+
+### 6. Use Metadata
+The `meta` field is used for:
+- Server information (server name, region, version)
+- User context (roles, permissions)
+- Message context (reply-to, thread-id)
+
+### 7. Thread-Safe Writes
+Use mutex/locks when writing to WebSocket from multiple threads.
+
+---
+
+
+## API Reference
+
+### Endpoints
+
+#### `GET /sync`
+WebSocket upgrade endpoint for mod/bot connections.
+
+**Query Parameters:**
+- `api_key` (required): Authentication token
+
+#### `GET /health`
+Health check endpoint.
+
+**Response:**
+```json
+{
+  "status": "healthy"
+}
+```
+
+---
+
+## Configuration
+
+Gateway configuration (`config.toml`):
+
+```toml
+[gateway]
+http_port = 3333          # WebSocket port
+websocket = "gateway"     # API key
+body_size = 1             # Max message size in MB
+queue_max = 8             # Messages per channel queue
+```
+
+---

bot.go

@@ -0,0 +1,253 @@
+//go:build simbot
+
+package main
+
+import (
+	"encoding/json"
+	"flag"
+	"fmt"
+	"log"
+	"math/rand"
+	"net/url"
+	"os"
+	"os/signal"
+	"sync"
+	"sync/atomic"
+	"syscall"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+const (
+	gatewayURL = "ws://localhost:3333/sync"
+	apiKey     = "gateway"
+	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 {
+	BotID string `json:"bot_id"`
+}
+
+type GatewayAck struct {
+	Status string `json:"status"`
+	Type   string `json:"type"`
+}
+
+type User struct {
+	ID   string `json:"id"`
+	Name string `json:"name"`
+}
+
+type Destination struct {
+	ID string `json:"channel_id,omitempty"`
+}
+
+type GatewayMessageIn struct {
+	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:"-"`
+}
+
+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() {
+	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)
+	signal.Notify(interrupt, os.Interrupt, syscall.SIGTERM)
+
+	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()
+
+	conn, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
+	if err != nil {
+		log.Fatalf("Failed to connect: %v", err)
+	}
+	defer conn.Close()
+
+	var writeMu sync.Mutex
+
+	conn.SetPingHandler(func(appData string) error {
+		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
+	})
+
+	bhs := BotHandshake{BotID: "discord-bot"}
+	data, err := json.Marshal(bhs)
+	if err != nil {
+		_ = conn.Close()
+		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)
+	}
+	writeMu.Unlock()
+
+	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{})
+
+	if msgType == websocket.TextMessage || msgType == websocket.BinaryMessage {
+		log.Printf("Raw handshake reply: %s", string(raw))
+		var ack GatewayAck
+		if err := json.Unmarshal(raw, &ack); err != nil {
+			log.Printf("Handshake reply is not JSON or unmarshal failed: %v", err)
+		} else {
+			log.Printf("Parsed ack: status=%q, type=%q", ack.Status, ack.Type)
+		}
+	} else {
+		log.Printf("Handshake reply type=%d", msgType)
+	}
+
+	done := make(chan struct{})
+
+	go func() {
+		defer close(done)
+		for {
+			msgType, 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 or read error: %v", err)
+				}
+				return
+			}
+			if msgType == websocket.TextMessage || msgType == websocket.BinaryMessage {
+				log.Printf("Received from gateway: %s", string(message))
+			}
+		}
+	}()
+
+	var msgCounter uint64 = 1
+
+	if *sendAfter > 0 {
+		go func() {
+			time.Sleep(*sendAfter)
+			msg := GatewayMessageIn{
+				MsgID: fmt.Sprintf("bot-msg-%06d", atomic.AddUint64(&msgCounter, 1)),
+				ID:    channelID,
+				Author: User{
+					ID:   *botID,
+					Name: "SimBot",
+				},
+				Content: *sendMsg,
+				Ts:      time.Now().UTC(),
+			}
+			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)
+			} else {
+				log.Printf("Sent bot->mod test message (channel=%s)", channelID)
+			}
+			_ = conn.SetWriteDeadline(time.Time{})
+			writeMu.Unlock()
+		}()
+	}
+
+	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)
+			}
+		}
+	}()
+
+	for {
+		select {
+		case <-done:
+			log.Println("Connection closed by server")
+			_ = conn.Close()
+			return
+		case <-interrupt:
+			log.Println("Interrupt received, closing connection...")
+			writeMu.Lock()
+			_ = conn.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
+			writeMu.Unlock()
+			select {
+			case <-done:
+			case <-time.After(time.Second):
+			}
+			_ = conn.Close()
+			return
+		}
+	}
+}

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 = 8

controller/controller.go

@@ -12,11 +12,8 @@ func NewGatewayController(cfg config.Config) GatewayController {
 		panic(err)
 	}
 
-	modHandler := ws.NewLoggingModHandler(wsl)
-	botHandler := ws.NewLoggingBotHandler(wsl)
-
 	return GatewayController{
-		Websocket:  ws.NewWebsocketGateway(cfg.Gateway, wsl, wCloseFn, modHandler, botHandler),
+		Websocket:  ws.NewWebsocketGateway(cfg.Gateway, wsl, wCloseFn),
 		HttpServer: HttpGateway{},
 	}
 }

controller/structs.go

@@ -1,8 +1,6 @@
 package controller
 
-import (
-	"homestead/homestead_gateway/ws"
-)
+import "homestead/homestead_gateway/ws"
 
 type GatewayController struct {
 	Websocket  *ws.WebsocketGateway

main.go

@@ -3,11 +3,20 @@ package main
 import (
 	"flag"
 	"homestead/homestead_gateway/controller"
+	"homestead/homestead_gateway/util"
 	"homestead/homestead_gateway/util/config"
+	"os"
+	"path"
 )
 
 func main() {
-	cfgPath := flag.String("config", "config.toml", "configuration file")
+	dir := util.GetPath()
+	if dir == "" {
+		dir, _ = os.Getwd()
+	}
+
+	file := path.Join(dir, "config.toml")
+	cfgPath := flag.String("config", file, "configuration file")
 	cfg, err := config.LoadConfig(*cfgPath)
 	if err != nil {
 		panic(err)
@@ -20,8 +29,4 @@ func main() {
 	}
 }
 
-/** TODO
-
-- queue for messages, both ways (Ack "queued" instead of "completed"), filled queue drops oldest entry
-
-*/
+// todo logs from exe not cwd

sim.go

@@ -1,11 +1,17 @@
+//go:build sim
+
 package main
 
 import (
 	"encoding/json"
+	"fmt"
 	"log"
+	"math/rand"
 	"net/url"
 	"os"
 	"os/signal"
+	"sync"
+	"sync/atomic"
 	"syscall"
 	"time"
 
@@ -13,9 +19,17 @@ import (
 )
 
 const (
-	gatewayURL = "ws://localhost:3333/push"
+	gatewayURL = "ws://localhost:3333/sync"
 	apiKey     = "gateway"
 	serverID   = "test-server-001"
+	// THE CHANNEL ID the mod says it serves. Must match gateway expectation.
+	channelID = "1444253682777587804"
+)
+
+// send interval range (random between minInterval and maxInterval)
+var (
+	minInterval = 500 * time.Millisecond
+	maxInterval = 5 * time.Second
 )
 
 type Handshake struct {
@@ -23,8 +37,10 @@ type Handshake struct {
 	Data json.RawMessage `json:"data"`
 }
 
+// ModHandshake now includes ChannelID
 type ModHandshake struct {
 	ServerID  string `json:"server_id"`
+	ChannelID string `json:"channel_id"`
 }
 
 type GatewayAck struct {
@@ -32,30 +48,32 @@ type GatewayAck struct {
 	Type   string `json:"type"`
 }
 
-type MinecraftUser struct {
+type User struct {
 	ID   string `json:"id"`
 	Name string `json:"name"`
 }
 
 type Destination struct {
-	ChannelID string `json:"channel_id"`
+	ID string `json:"channel_id,omitempty"`
 }
 
-type GatewayModMessageIn struct {
-	MsgID       string                 `json:"msg_id"`
-	Server      string                 `json:"server"`
-	Destination Destination            `json:"destination"`
-	Author      MinecraftUser          `json:"author"`
-	Content     string                 `json:"content"`
-	Meta        map[string]interface{} `json:"meta,omitempty"`
-	Ts          string                 `json:"ts,omitempty"`
+type GatewayMessageIn struct {
+	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)
 }
 
 func main() {
+	rand.Seed(time.Now().UnixNano())
+
 	interrupt := make(chan os.Signal, 1)
 	signal.Notify(interrupt, os.Interrupt, syscall.SIGTERM)
 
-	// Build WebSocket URL with API key
 	u, err := url.Parse(gatewayURL)
 	if err != nil {
 		log.Fatalf("Failed to parse URL: %v", err)
@@ -66,7 +84,6 @@ func main() {
 
 	log.Printf("Connecting to %s", u.String())
 
-	// Connect to WebSocket
 	conn, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
 	if err != nil {
 		log.Fatalf("Failed to connect: %v", err)
@@ -75,19 +92,23 @@ func main() {
 
 	log.Println("Connected to gateway")
 
-	// Set up ping handler - respond to pings from server
+	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
 	})
 
-	// Send handshake
-	modHS := ModHandshake{ServerID: serverID}
+	modHS := ModHandshake{
+		ServerID:  serverID,
+		ChannelID: channelID,
+	}
 	modHSData, err := json.Marshal(modHS)
 	if err != nil {
 		log.Fatalf("Failed to marshal mod handshake: %v", err)
@@ -98,24 +119,22 @@ 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
 	var ack GatewayAck
 	if err := conn.ReadJSON(&ack); err != nil {
 		log.Fatalf("Failed to read acknowledgment: %v", err)
 	}
+	log.Printf("Received acknowledgment: status=%q, type=%q", ack.Status, ack.Type)
 
-	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 {
@@ -124,66 +143,115 @@ 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
 			}
-
-			// Only log text/binary messages (ping/pong handled by handlers)
-			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(2 * time.Second)
-	testMsg := GatewayModMessageIn{
-		MsgID:  "test-msg-001",
-		Server: serverID,
+	var msgCounter uint64 = 1
+
+	func() {
+		testMsg := GatewayMessageIn{
+			MsgID: fmt.Sprintf("test-msg-%06d", atomic.AddUint64(&msgCounter, 1)),
+			ID:    serverID,
 			Destination: Destination{
-			ChannelID: "123456789",
+				ID: channelID,
 			},
-		Author: MinecraftUser{
+			Author: User{
 				ID:   "player-uuid-123",
 				Name: "TestPlayer",
 			},
 			Content: "Hello from simulated mod!",
-		Ts:      time.Now().UTC().Format(time.RFC3339),
+			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.")
 
-	// Wait for interrupt or connection close
 	for {
 		select {
 		case <-done:
-			log.Println("Connection closed")
+			log.Println("Connection read loop closed, exiting")
 			return
 		case <-interrupt:
 			log.Println("Interrupt received, closing connection...")
 
-			// Send close message
+			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/cache/cache.go

@@ -1,91 +0,0 @@
-package cache
-
-func NewCache() *Cache {
-	c := &Cache{}
-	c.state.Store(&mappings{
-		s2c: make(map[string]string),
-		c2s: make(map[string]string),
-	})
-
-	return c
-}
-
-func (c *Cache) Set(serverId, channelId string) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	current := c.state.Load()
-	next := current.clone()
-
-	if oldCh, ok := next.s2c[serverId]; ok && oldCh != channelId {
-		delete(next.c2s, oldCh)
-	}
-	if oldSrv, ok := next.c2s[channelId]; ok && oldSrv != serverId {
-		delete(next.s2c, oldSrv)
-	}
-
-	next.s2c[serverId] = channelId
-	next.c2s[channelId] = serverId
-
-	c.state.Store(next)
-}
-
-func (c *Cache) GetByServerId(serverId string) (string, bool) {
-	m := c.state.Load()
-	val, ok := m.s2c[serverId]
-	return val, ok
-}
-
-func (c *Cache) GetByChannelId(channelId string) (string, bool) {
-	m := c.state.Load()
-	val, ok := m.c2s[channelId]
-	return val, ok
-}
-
-func (c *Cache) RemoveByServerId(serverId string) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	current := c.state.Load()
-	if _, ok := current.s2c[serverId]; !ok {
-		return
-	}
-
-	next := current.clone()
-	if channelId, ok := next.s2c[serverId]; ok {
-		delete(next.s2c, serverId)
-		delete(next.c2s, channelId)
-	}
-	c.state.Store(next)
-}
-
-func (c *Cache) RemoveByChannelId(channelId string) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	current := c.state.Load()
-	if _, ok := current.c2s[channelId]; !ok {
-		return
-	}
-
-	next := current.clone()
-	if serverId, ok := next.c2s[channelId]; ok {
-		delete(next.c2s, channelId)
-		delete(next.s2c, serverId)
-	}
-	c.state.Store(next)
-}
-
-func (m *mappings) clone() *mappings {
-	newM := &mappings{
-		s2c: make(map[string]string, len(m.s2c)),
-		c2s: make(map[string]string, len(m.c2s)),
-	}
-	for k, v := range m.s2c {
-		newM.s2c[k] = v
-	}
-	for k, v := range m.c2s {
-		newM.c2s[k] = v
-	}
-	return newM
-}

util/cache/conn_cache.go

@@ -1,88 +0,0 @@
-package cache
-
-import (
-	"github.com/gorilla/websocket"
-)
-
-func NewConnectionCache() *ConnectionCache {
-	c := &ConnectionCache{}
-	c.state.Store(&connectionMappings{
-		id2c: make(map[string]*conn),
-	})
-
-	return c
-}
-
-func (c *ConnectionCache) Set(id string, connection *websocket.Conn, meta *ConnectionMetaData) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	current := c.state.Load()
-	next := current.clone()
-
-	next.id2c[id] = &conn{
-		connection: connection,
-		meta:       meta,
-	}
-
-	c.state.Store(next)
-}
-
-func (c *ConnectionCache) GetById(id string) (*websocket.Conn, *ConnectionMetaData, bool) {
-	m := c.state.Load()
-	if conn, ok := m.id2c[id]; ok {
-		return conn.connection, conn.meta, true
-	}
-
-	return nil, nil, false
-}
-
-func (c *ConnectionCache) RemoveById(id string) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	current := c.state.Load()
-	if _, ok := current.id2c[id]; !ok {
-		return
-	}
-
-	next := current.clone()
-	delete(next.id2c, id)
-
-	c.state.Store(next)
-}
-
-func (c *ConnectionCache) Range(fn func(id string, connection *websocket.Conn, meta *ConnectionMetaData)) {
-	m := c.state.Load()
-	for id, conn := range m.id2c {
-		fn(id, conn.connection, conn.meta)
-	}
-}
-
-func (c *ConnectionCache) GetAllConnections() []*websocket.Conn {
-	m := c.state.Load()
-	conns := make([]*websocket.Conn, 0, len(m.id2c))
-	for _, conn := range m.id2c {
-		conns = append(conns, conn.connection)
-	}
-	return conns
-}
-
-func (c *ConnectionCache) Clear() {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	c.state.Store(&connectionMappings{
-		id2c: make(map[string]*conn),
-	})
-}
-
-func (cm *connectionMappings) clone() *connectionMappings {
-	newCM := &connectionMappings{
-		id2c: make(map[string]*conn, len(cm.id2c)),
-	}
-	for k, v := range cm.id2c {
-		newCM.id2c[k] = v
-	}
-	return newCM
-}

util/cache/structs.go

@@ -1,37 +0,0 @@
-package cache
-
-import (
-	"sync"
-	"sync/atomic"
-
-	"github.com/gorilla/websocket"
-)
-
-type Cache struct {
-	mu    sync.Mutex
-	state atomic.Pointer[mappings]
-}
-
-type ConnectionCache struct {
-	mu    sync.Mutex
-	state atomic.Pointer[connectionMappings]
-}
-
-type mappings struct {
-	s2c map[string]string
-	c2s map[string]string
-}
-
-type connectionMappings struct {
-	id2c map[string]*conn
-}
-
-type ConnectionMetaData struct {
-	ConnectionType string // "mod" or "bot"
-	ID             string // server_id or bot_id for logging
-}
-
-type conn struct {
-	connection *websocket.Conn
-	meta       *ConnectionMetaData
-}

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"`
-}

util/logger/log.go

@@ -3,6 +3,7 @@ package logger
 import (
 	"context"
 	"fmt"
+	"homestead/homestead_gateway/util"
 	"homestead/homestead_gateway/util/config"
 	"log/slog"
 	"os"
@@ -23,6 +24,7 @@ func New(id string, cfg config.LogConfig) (*slog.Logger, func() error, error) {
 		cfg.Rotation = 7
 	}
 
+	cfg.Directory = util.NormalizeLogPath(cfg.Directory)
 	console := slog.NewTextHandler(&prefixWriter{inner: os.Stderr, prefix: []byte("[" + id + "] "), startLine: true}, &slog.HandlerOptions{AddSource: true, Level: cfg.Level})
 	router := newFileRouter(cfg.Directory, cfg.Rotation, id)
 	root := slogmulti.Fanout(console, router)

util/util.go

@@ -0,0 +1,51 @@
+package util
+
+import (
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+func GetPath() string {
+	exe, err := os.Executable()
+	if err != nil {
+		return ""
+	}
+
+	exe, err = filepath.EvalSymlinks(exe)
+	if err != nil {
+	}
+
+	return filepath.Dir(exe)
+}
+
+func NormalizePath(path string) string {
+	return filepath.Clean(filepath.FromSlash(path))
+}
+
+func NormalizeLogPath(path string) string {
+	if filepath.IsAbs(path) {
+		return path
+	}
+
+	return NormalizePath(filepath.Join(GetPath(), path))
+}
+
+//
+
+func CloseConnWithControlMessage(conn *websocket.Conn, typ int, text string) {
+	_ = conn.SetWriteDeadline(time.Now().Add(time.Second))
+	_ = conn.WriteControl(
+		typ, websocket.FormatCloseMessage(typ, text), time.Now().Add(time.Second),
+	)
+	_ = conn.Close()
+}
+
+func CloseConn(conn *websocket.Conn) {
+	CloseConnWithControlMessage(
+		conn, websocket.CloseNormalClosure,
+		"Disconnecting.",
+	)
+}

ws/handlers.go

@@ -2,14 +2,13 @@ package ws
 
 import (
 	"encoding/json"
-	"homestead/homestead_gateway/util/cache"
 	"net/http"
 	"time"
 
 	"github.com/gorilla/websocket"
 )
 
-func (wsg *WebsocketGateway) handlePush(w http.ResponseWriter, r *http.Request) {
+func (wsg *WebsocketGateway) handleSync(w http.ResponseWriter, r *http.Request) {
 	conn, err := wsg.validateAndUpgradeConnection(w, r)
 	if err != nil {
 		return
@@ -29,80 +28,81 @@ func (wsg *WebsocketGateway) handlePush(w http.ResponseWriter, r *http.Request)
 
 	typ, data, err := conn.ReadMessage()
 	if err != nil {
-		wsg.sendWebsocketError(conn, "Internal Server Error", 500)
+		wsg.sendWebsocketError(conn, "Internal Server Error", 500, true)
 		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.sendWebsocketError(conn, "Initial message must be a handshake.", 400, true)
 		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.sendWebsocketError(conn, "Malformed handshake.", 400, true)
 		wsg.logger.Warn("Malformed handshake.", "remote", conn.RemoteAddr().String(), "err", err)
 		return
 	}
 
-	meta := cache.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.sendWebsocketError(conn, "Malformed mod handshake.", 400, true)
 			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 {
+		if mhs.ServerID == "" || mhs.ChannelID == "" {
+			wsg.sendWebsocketError(conn, "Malformed mod handshake.", 400, true)
 			return
 		}
 
-		wsg.registerConn(conn, meta)
-		wsg.logger.Info("Mod connected via Websocket.", "remote", conn.RemoteAddr().String(), "server_id", mhs.ServerID)
+		if !wsg.registerConn(conn, "mod", mhs.ChannelID, mhs.ServerID) {
+			wsg.sendWebsocketError(conn, "Failed to register mod.", 500, true)
+			return
+		}
 
-		go wsg.modReadLoop(conn, meta) // replace with external handler mayhaps
+		_ = wsg.sendWebsocketResponse(conn, GatewayAck{Status: "connected", Type: "mod"})
+		wsg.registry.FlushChannelWithSender(mhs.ChannelID, wsg.flush)
+
+		wsg.logger.Info("Mod connected via Websocket.", "remote", conn.RemoteAddr().String(), "id", mhs.ServerID)
+		go wsg.read(conn, "mod", mhs.ChannelID)
 
 	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)
+			wsg.sendWebsocketError(conn, "Malformed bot handshake.", 400, true)
 			return
 		}
 
-		meta.ID = bhs.BotID
-
-		if err = wsg.sendWebsocketResponse(conn, GatewayAck{Status: "connected", Type: "bot"}); err != nil {
+		if bhs.BotID == "" {
+			wsg.sendWebsocketError(conn, "Malformed bot handshake.", 400, true)
 			return
 		}
 
-		wsg.registerConn(conn, meta)
-		wsg.logger.Info("Bot connected via Websocket.", "remote", conn.RemoteAddr().String(), "bot_id", bhs.BotID)
+		if !wsg.registerConn(conn, "bot", "", "") {
+			wsg.sendWebsocketError(conn, "Bot already connected.", 409, true)
+			return
+		}
 
-		go wsg.botReadLoop(conn, meta) // replace with external handler mayhaps
+		_ = wsg.sendWebsocketResponse(conn, GatewayAck{Status: "connected", Type: "bot"})
+		wsg.registry.FlushAllToBotWithSender(wsg.flush)
+
+		wsg.logger.Info("Bot connected via Websocket.", "remote", conn.RemoteAddr().String(), "id", bhs.BotID)
+		go wsg.read(conn, "bot", "")
 
 	default:
-		wsg.sendWebsocketError(conn, "Unknown handshake.", 400)
+		wsg.sendWebsocketError(conn, "Unknown handshake.", 400, true)
 		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/registry.go

@@ -0,0 +1,252 @@
+package ws
+
+import (
+	"fmt"
+	"homestead/homestead_gateway/util"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+func (q *BoundedQueue) Enqueue(m GatewayMessageOut) bool {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	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
+}
+
+func (q *BoundedQueue) PopAll() []GatewayMessageOut {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	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
+}
+
+func (q *BoundedQueue) Len() int {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	return q.length
+}
+
+//
+
+func (r *Registry) getOrCreate(channel string) *ChannelEntry {
+	r.mu.RLock()
+	e := r.entries[channel]
+	r.mu.RUnlock()
+	if e != nil {
+		return e
+	}
+
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if e = r.entries[channel]; e == nil {
+		e = newChannelEntry(channel, r.queueCap)
+		r.entries[channel] = 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)
+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 {
+		util.CloseConn(e.Mod.Conn)
+	}
+
+	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
+}
+
+// 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.UnregisterBot()
+	}
+
+	r.bot = &ConnWrapper{Conn: conn, LastSeen: time.Now()}
+	r.botMu.Unlock()
+}
+
+func (r *Registry) UnregisterMod(channelID string) {
+	r.mu.RLock()
+	e := r.entries[channelID]
+	r.mu.RUnlock()
+	if e == nil {
+		return
+	}
+
+	e.mu.Lock()
+	modConn := e.Mod
+	e.Mod = nil
+	e.mu.Unlock()
+
+	if modConn != nil && modConn.Conn != nil {
+		util.CloseConn(modConn.Conn)
+	}
+}
+
+func (r *Registry) UnregisterBot() {
+	r.botMu.Lock()
+	botConn := r.bot
+	r.bot = nil
+	r.botMu.Unlock()
+
+	if botConn != nil && botConn.Conn != nil {
+		util.CloseConn(botConn.Conn)
+	}
+}
+
+func (r *Registry) Send(channelID string, out GatewayMessageOut, sendOverConn func(*websocket.Conn, GatewayMessageOut) error) (delivered bool, queued bool, err error) {
+	if out.Type == "mod" {
+		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
+			}
+			r.UnregisterBot()
+		}
+
+		e := r.getOrCreate(channelID)
+		e.mu.Lock()
+		enq := e.Queue.Enqueue(out)
+		e.mu.Unlock()
+
+		if !enq {
+			return false, false, fmt.Errorf("queue disabled")
+		}
+		return false, true, nil
+	}
+
+	e := r.getOrCreate(channelID)
+	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
+		}
+		r.UnregisterMod(channelID)
+	}
+
+	e.mu.Lock()
+	enq := e.Queue.Enqueue(out)
+	e.mu.Unlock()
+
+	if !enq {
+		return false, false, fmt.Errorf("queue disabled")
+	}
+	return false, true, nil
+}
+
+//
+
+func (r *Registry) FlushChannelWithSender(channelID string, sendOverConn func(*websocket.Conn, GatewayMessageOut) error) {
+	r.mu.RLock()
+	e := r.entries[channelID]
+	r.mu.RUnlock()
+	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()
+
+	for _, m := range msgs {
+		if err := sendOverConn(modConn, m); err != nil {
+			// if send fails, re-enqueue (best-effort), drop-oldest logic applies
+			e.mu.Lock()
+			_ = e.Queue.Enqueue(m)
+			e.mu.Unlock()
+		}
+	}
+}
+
+func (r *Registry) FlushAllToBotWithSender(sendOverConn func(*websocket.Conn, GatewayMessageOut) error) {
+	r.botMu.Lock()
+	b := r.bot
+	r.botMu.Unlock()
+	if b == nil || b.Conn == nil {
+		return
+	}
+
+	r.mu.RLock()
+	entries := make([]*ChannelEntry, 0, len(r.entries))
+	for _, e := range r.entries {
+		entries = append(entries, e)
+	}
+	r.mu.RUnlock()
+
+	for _, e := range entries {
+		e.mu.Lock()
+		msgs := e.Queue.PopAll()
+		e.mu.Unlock()
+
+		if len(msgs) == 0 {
+			continue
+		}
+		for _, m := range msgs {
+			if err := sendOverConn(b.Conn, m); err != nil {
+				e.mu.Lock()
+				_ = e.Queue.Enqueue(m)
+				e.mu.Unlock()
+			}
+		}
+	}
+}

ws/structs.go

@@ -2,8 +2,8 @@ package ws
 
 import (
 	"encoding/json"
-	"homestead/homestead_gateway/util/cache"
 	"log/slog"
+	"sync"
 	"time"
 
 	"github.com/gorilla/websocket"
@@ -16,74 +16,74 @@ type WebsocketGateway struct {
 
 	upgrader websocket.Upgrader
 
-	cache   *cache.Cache
-	botConn *websocket.Conn
-	conns   *cache.ConnectionCache
-
-	modHandler ModHandler
-	botHandler BotHandler
+	registry *Registry
 
 	logger  *slog.Logger
 	closeFn func() error
 }
 
-type MinecraftUser struct {
-	ID   string `json:"id"`
-	Name string `json:"name"`
+//
+
+type Registry struct {
+	mu       sync.RWMutex
+	entries  map[string]*ChannelEntry
+	queueCap int
+
+	botMu sync.Mutex
+	bot   *ConnWrapper
 }
 
-type DiscordUser struct {
+type ConnWrapper struct {
+	Conn     *websocket.Conn
+	ServerID string // set for mods (the server_id)
+	LastSeen time.Time
+}
+
+type BoundedQueue struct {
+	mu       sync.Mutex
+	buf      []GatewayMessageOut
+	start    int
+	length   int
+	capacity int
+}
+
+type ChannelEntry struct {
+	mu      sync.Mutex
+	Channel string
+	Mod     *ConnWrapper
+	Queue   *BoundedQueue
+}
+
+//
+
+type User struct {
 	ID   string `json:"id"`
 	Name string `json:"name"`
 }
 
 type Destination struct {
-	ChannelID string `json:"channel_id"`
+	ID string `json:"channel_id,omitempty"`
 }
 
-// GatewayModMessageIn : Mod -> Gateway -> Bot
-type GatewayModMessageIn struct {
+type GatewayMessageIn struct {
+	Type        string
+	ID          string                 `json:"id"`
 	MsgID       string                 `json:"msg_id"`
-	Server      string                 `json:"server"`
-	Destination Destination            `json:"destination"`
-	Author      MinecraftUser          `json:"author"`
+	Destination Destination            `json:"destination,omitempty"`
+	Author      User                   `json:"author"`
 	Content     string                 `json:"content"`
 	Meta        map[string]interface{} `json:"meta,omitempty"`
-	Ts          string                 `json:"ts,omitempty"`
-	ReceivedAt  time.Time              `json:"-"` // ReceivedAt is populated by gateway (not from mod)
+	Ts          time.Time              `json:"ts,omitempty"`
+	ReceivedAt  time.Time              `json:"-"`
 }
 
-// GatewayBotMessageIn : Bot -> Gateway -> Mod
-type GatewayBotMessageIn struct {
-	MsgID      string                 `json:"msg_id"`
-	ChannelID  string                 `json:"channel_id"`
-	Author     DiscordUser            `json:"author"`
+type GatewayMessageOut struct {
+	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"`
-	Ts         string                 `json:"ts,omitempty"`
-	ReceivedAt time.Time              `json:"-"` // ReceivedAt is populated by gateway (not from bot)
-}
-
-// GatewayModMessageOut : Gateway<GatewayModMessageIn> -> Bot
-type GatewayModMessageOut struct {
-	Type        string                 `json:"type"` // "mod"
-	ChannelID   string                 `json:"channel_id"`
-	Author      MinecraftUser          `json:"author"`
-	Content     string                 `json:"content"`
-	Meta        map[string]interface{} `json:"meta,omitempty"`
-	Ts          string                 `json:"ts,omitempty"`
-	ReceivedAt  time.Time              `json:"received_at"`
-	ForwardedAt time.Time              `json:"forwarded_at"`
-}
-
-// GatewayBotMessageOut : Gateway<GatewayBotMessageIn> -> Mod
-type GatewayBotMessageOut struct {
-	Type        string                 `json:"type"` // "bot"
-	ChannelID   string                 `json:"channel_id"`
-	Author      DiscordUser            `json:"author"`
-	Content     string                 `json:"content"`
-	Meta        map[string]interface{} `json:"meta,omitempty"`
-	Ts          string                 `json:"ts,omitempty"`
+	Ts          time.Time              `json:"ts,omitempty"`
 	ReceivedAt  time.Time              `json:"received_at"`
 	ForwardedAt time.Time              `json:"forwarded_at"`
 }
@@ -93,6 +93,8 @@ type GatewayAck struct {
 	Type   string `json:"type"`
 }
 
+//
+
 type Handshake struct {
 	Type string          `json:"type"` // "mod" or "bot"
 	Data json.RawMessage `json:"data"`
@@ -100,16 +102,9 @@ type Handshake struct {
 
 type ModHandshake struct {
 	ServerID  string `json:"server_id"`
+	ChannelID string `json:"channel_id"`
 }
 
 type BotHandshake struct {
 	BotID string `json:"bot_id"`
 }
-
-type ModHandler interface {
-	Handle(conn *websocket.Conn, msg GatewayModMessageIn) error
-}
-
-type BotHandler interface {
-	Handle(conn *websocket.Conn, msg GatewayBotMessageIn) error
-}

ws/temp.go

@@ -1,71 +0,0 @@
-package ws
-
-import (
-	"encoding/json"
-	"log/slog"
-	"time"
-
-	"github.com/gorilla/websocket"
-)
-
-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(conn *websocket.Conn, 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(conn *websocket.Conn, 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/util.go

@@ -4,9 +4,9 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"homestead/homestead_gateway/util/cache"
-	"log/slog"
+	"homestead/homestead_gateway/util"
 	"net/http"
+	"strings"
 	"time"
 
 	"github.com/gorilla/websocket"
@@ -36,6 +36,11 @@ func (wsg *WebsocketGateway) deafen(srv *http.Server) {
 
 // responses
 
+func (wsg *WebsocketGateway) flush(c *websocket.Conn, m GatewayMessageOut) error {
+	_ = c.SetWriteDeadline(time.Now().Add(5 * time.Second))
+	return c.WriteJSON(m)
+}
+
 func (wsg *WebsocketGateway) sendHttpError(w http.ResponseWriter, message string, code int) {
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(code)
@@ -47,10 +52,12 @@ func (wsg *WebsocketGateway) sendWebsocketPing(conn *websocket.Conn) {
 	_ = conn.WriteMessage(websocket.PingMessage, nil)
 }
 
-func (wsg *WebsocketGateway) sendWebsocketError(conn *websocket.Conn, message string, code int) {
+func (wsg *WebsocketGateway) sendWebsocketError(conn *websocket.Conn, message string, code int, close bool) {
 	_ = conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
 	_ = conn.WriteJSON(map[string]interface{}{"message": message, "code": code})
-	_ = conn.Close()
+	if close {
+		util.CloseConn(conn)
+	}
 }
 
 func (wsg *WebsocketGateway) sendWebsocketResponse(conn *websocket.Conn, content interface{}) error {
@@ -58,7 +65,7 @@ func (wsg *WebsocketGateway) sendWebsocketResponse(conn *websocket.Conn, content
 
 	if err := conn.WriteJSON(content); err != nil {
 		wsg.logger.Error("Failed to respond to connection.", "remote", conn.RemoteAddr().String(), "err", err)
-		_ = conn.Close()
+		util.CloseConnWithControlMessage(conn, websocket.CloseAbnormalClosure, "Connection error.")
 		return err
 	}
 
@@ -92,39 +99,106 @@ func (wsg *WebsocketGateway) validateApiKey(r *http.Request) bool {
 	return !(apiKey == "" || apiKey != wsg.apiKey)
 }
 
-func writeJSONSafe(c *websocket.Conn, v interface{}) error {
-	_ = c.SetWriteDeadline(time.Now().Add(5 * time.Second))
-	if err := c.WriteJSON(v); err != nil {
-		// caller handles logging
-		return err
-	}
-	return nil
-}
-
-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("http request", "remote", r.RemoteAddr, "method", r.Method, "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 (wsg *WebsocketGateway) registerConn(conn *websocket.Conn, meta cache.ConnectionMetaData) {
-	wsg.conns.Set(meta.ID, conn, &meta)
+func (wsg *WebsocketGateway) registerConn(conn *websocket.Conn, typ, channelId, serverId string) bool {
+	if typ == "bot" {
+		wsg.registry.botMu.Lock()
+		if wsg.registry.bot != nil && wsg.registry.bot.Conn != nil {
+			wsg.registry.botMu.Unlock()
+			return false
+		}
+		wsg.registry.botMu.Unlock()
+
+		wsg.registry.RegisterBot(conn)
+		return true
+	}
+
+	wsg.registry.RegisterMod(channelId, serverId, conn)
+	return true
 }
 
-func (wsg *WebsocketGateway) unregisterConn(conn *websocket.Conn, meta cache.ConnectionMetaData) {
-	wsg.conns.RemoveById(meta.ID)
-	_ = conn.Close()
+func (wsg *WebsocketGateway) unregisterConn(typ, channelId string) {
+	if typ == "bot" {
+		wsg.registry.UnregisterBot()
+		return
+	}
+
+	wsg.registry.UnregisterMod(channelId)
 }
 
 func (wsg *WebsocketGateway) closeAll() {
 	wsg.logger.Info("Closing all websocket connections.")
 
-	wsg.conns.Range(func(id string, c *websocket.Conn, meta *cache.ConnectionMetaData) {
-		_ = c.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "Shutting down."), time.Now().Add(time.Second))
-		_ = c.Close()
+	wsg.registry.UnregisterBot()
+
+	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),
+		queueCap: queueCap,
+	}
+}
+
+func NewBoundedQueue(cap int) *BoundedQueue {
+	if cap <= 0 {
+		cap = 128
+	}
+	return &BoundedQueue{buf: make([]GatewayMessageOut, cap), capacity: cap}
+}
+
+func newChannelEntry(channel string, cap int) *ChannelEntry {
+	return &ChannelEntry{
+		Channel: channel,
+		Queue:   NewBoundedQueue(cap),
+	}
+}
+
+//
+
+func (m *GatewayMessageIn) Validate() error {
+	if strings.TrimSpace(m.ID) == "" {
+		return errors.New("id missing")
+	}
+	if strings.TrimSpace(m.MsgID) == "" {
+		return errors.New("msg_id missing")
+	}
+	if strings.TrimSpace(m.Author.ID) == "" {
+		return errors.New("author.id missing")
+	}
+	if strings.TrimSpace(m.Content) == "" {
+		return errors.New("content missing")
+	}
+
+	if m.Type == "mod" && strings.TrimSpace(m.Destination.ID) == "" {
+		return errors.New("destination.channel_id missing")
+	}
+
+	return nil
+}
+
+func (c *ConnWrapper) Alive() bool { return c != nil && c.Conn != nil }

ws/validate.go

@@ -1,38 +0,0 @@
-package ws
-
-import (
-	"errors"
-	"strings"
-)
-
-func (m *GatewayModMessageIn) Validate() error {
-	if strings.TrimSpace(m.MsgID) == "" {
-		return errors.New("msg_id missing")
-	}
-	if strings.TrimSpace(m.Server) == "" {
-		return errors.New("server missing")
-	}
-	if strings.TrimSpace(m.Author.ID) == "" {
-		return errors.New("author.id missing")
-	}
-	if strings.TrimSpace(m.Content) == "" {
-		return errors.New("content missing")
-	}
-	return nil
-}
-
-func (m *GatewayBotMessageIn) Validate() error {
-	if strings.TrimSpace(m.MsgID) == "" {
-		return errors.New("msg_id missing")
-	}
-	if strings.TrimSpace(m.ChannelID) == "" {
-		return errors.New("channel_id missing")
-	}
-	if strings.TrimSpace(m.Author.ID) == "" {
-		return errors.New("author missing")
-	}
-	if strings.TrimSpace(m.Content) == "" {
-		return errors.New("content missing")
-	}
-	return nil
-}

ws/websocket.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"homestead/homestead_gateway/util/cache"
 	"homestead/homestead_gateway/util/config"
 	"log/slog"
 	"net"
@@ -16,23 +15,14 @@ import (
 	"github.com/gorilla/websocket"
 )
 
-func NewWebsocketGateway(cfg config.GatewayConfig, logger *slog.Logger, closefn func() error, modH ModHandler, botH BotHandler) *WebsocketGateway {
+func NewWebsocketGateway(cfg config.GatewayConfig, logger *slog.Logger, closefn func() error) *WebsocketGateway {
 	return &WebsocketGateway{
 		logger:        logger,
 		closeFn:       closefn,
-		modHandler: modH,
-		botHandler: botH,
 		port:          cfg.HttpPort,
 		apiKey:        cfg.Websocket,
-		cache:      cache.NewCache(),
-		conns:      cache.NewConnectionCache(),
-		upgrader: websocket.Upgrader{
-			ReadBufferSize:  1024,
-			WriteBufferSize: 1024,
-			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,
 	}
 }
@@ -46,14 +36,12 @@ func (wsg *WebsocketGateway) Start() error {
 
 func (wsg *WebsocketGateway) Serve(ctx context.Context, listenAddr string) error {
 	mux := http.NewServeMux()
-	mux.HandleFunc("/push", wsg.handlePush)
-	mux.HandleFunc("/ready", wsg.handleReady)
+	mux.HandleFunc("/sync", wsg.handleSync)
 	mux.HandleFunc("/health", wsg.handleHealth)
-	mux.HandleFunc("/register", wsg.handleRegister)
 
 	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)
@@ -71,29 +59,33 @@ func (wsg *WebsocketGateway) Serve(ctx context.Context, listenAddr string) error
 
 //
 
-func (wsg *WebsocketGateway) modReadLoop(conn *websocket.Conn, meta cache.ConnectionMetaData) {
+func (wsg *WebsocketGateway) read(conn *websocket.Conn, _type, channelId string) {
 	defer func() {
-		wsg.unregisterConn(conn, meta)
-		wsg.logger.Info("Mod-Client disconnected.", "remote", conn.RemoteAddr().String(), "server_id", meta.ID)
+		wsg.unregisterConn(_type, channelId)
+		wsg.logger.Info("Client disconnected.", "remote", conn.RemoteAddr().String())
 	}()
 
 	ticker := time.NewTicker(30 * time.Second)
+	ctx, cancel := context.WithCancel(context.Background())
 	defer ticker.Stop()
+	defer cancel()
 
 	go func() {
-		for range ticker.C {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-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("unexpected mod close", "server_id", meta.ID, "err", err)
-			} else {
-				wsg.logger.Debug("mod read error", "server_id", meta.ID, "err", err)
+				wsg.logger.Error("Client unexpectedly closed the connection.", "err", err)
 			}
 			return
 		}
@@ -102,88 +94,59 @@ func (wsg *WebsocketGateway) modReadLoop(conn *websocket.Conn, meta cache.Connec
 			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)
+		ts := time.Now().UTC()
+		var message GatewayMessageIn
+		if err := json.Unmarshal(data, &message); err != nil {
+			wsg.sendWebsocketError(conn, "Malformed message.", 400, false)
+			wsg.logger.Warn("Received malformed message json from client.", "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)
+		message.Type = _type
+		message.ReceivedAt = ts
+		if err := message.Validate(); err != nil {
+			wsg.sendWebsocketError(conn, "Malformed message.", 400, false)
+			wsg.logger.Warn("Received malformed message json from client; validation failed.", "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)
-		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)
+		var outID string
+		if _type == "mod" {
+			outID = channelId
 		} else {
-				wsg.logger.Debug("bot read error", "bot_id", meta.ID, "err", err)
-			}
-			return
+			outID = message.ID
 		}
 
-		if typ != websocket.TextMessage && typ != websocket.BinaryMessage {
+		out := GatewayMessageOut{
+			Type:        message.Type,
+			ID:          outID,
+			Author:      message.Author,
+			Content:     message.Content,
+			Meta:        message.Meta,
+			Ts:          message.Ts,
+			ReceivedAt:  message.ReceivedAt,
+			ForwardedAt: time.Now().UTC(),
+		}
+
+		delivered, queued, err := wsg.registry.Send(outID, out, wsg.flush)
+
+		if err != nil {
+			wsg.logger.Error("Registry queue/delivery error.", "remote", conn.RemoteAddr().String(), "err", err)
+			_ = wsg.sendWebsocketResponse(conn, GatewayAck{Status: "failed", Type: message.Type})
 			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)
+		if delivered {
+			_ = wsg.sendWebsocketResponse(conn, GatewayAck{Status: "completed", Type: message.Type})
 			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)
+		if queued {
+			_ = wsg.sendWebsocketResponse(conn, GatewayAck{Status: "queued", Type: message.Type})
 			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
-		}
+		_ = wsg.sendWebsocketResponse(conn, GatewayAck{Status: "failed", Type: message.Type})
 
-		_ = writeJSONSafe(conn, map[string]string{"status": "ok"})
 	}
 }