Phase 1 MVP: gateway, CRUD handlers, frontend components

Backend: - WebSocket gateway (hub, client, events) with fanout broadcast - Server CRUD handlers (create, list, get, update, delete) - Channel CRUD handlers (create, list, get, update, delete) - Message CRUD handlers (list with cursor pagination, create, update, delete) - cmd/migrate standalone migration CLI (up/down) - cmd/server wired to all handlers + WebSocket + static file serving Frontend: - Zustand stores: auth, server, channel, message, websocket - API client with fetch wrapper - Terminal-styled components: Layout, LoginForm, ChatArea, ChannelList, ServerBar, MemberList - React Router with login and main routes - Gruvbox dark palette throughout Ops: - Docker Compose with app service (multi-stage build) - Caddyfile with WebSocket upgrade support - Makefile for common tasks
2026-06-26 14:47:29 -04:00
parent aa7854aee2
commit bb5a56816b
32 changed files with 2310 additions and 339 deletions
@@ -0,0 +1,154 @@
+package gateway
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"log/slog"
+	"net/http"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+const (
+	writeWait      = 10 * time.Second
+	pongWait       = 60 * time.Second
+	pingPeriod     = (pongWait * 9) / 10
+	maxMessageSize = 4096
+)
+
+var upgrader = websocket.Upgrader{
+	ReadBufferSize:  1024,
+	WriteBufferSize: 1024,
+	CheckOrigin:     func(r *http.Request) bool { return true },
+}
+
+// Client is a middleman between the websocket connection and the hub.
+type Client struct {
+	Hub    *Hub
+	Conn   *websocket.Conn
+	UserID string
+	send   chan []byte
+}
+
+// readPump pumps messages from the websocket connection to the hub.
+func (c *Client) readPump() {
+	defer func() {
+		c.Hub.Unregister(c)
+		c.Conn.Close()
+	}()
+
+	c.Conn.SetReadLimit(maxMessageSize)
+	c.Conn.SetReadDeadline(time.Now().Add(pongWait))
+	c.Conn.SetPongHandler(func(string) error {
+		c.Conn.SetReadDeadline(time.Now().Add(pongWait))
+		return nil
+	})
+
+	for {
+		_, raw, err := c.Conn.ReadMessage()
+		if err != nil {
+			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
+				c.Hub.logger.Warn("websocket read error", "error", err, "user_id", c.UserID)
+			}
+			break
+		}
+
+		var event Event
+		if err := json.Unmarshal(raw, &event); err != nil {
+			c.Hub.logger.Warn("invalid event JSON", "error", err, "user_id", c.UserID)
+			continue
+		}
+
+		// Handle client-sent events (e.g. TYPING_START, PRESENCE_UPDATE)
+		// For now, broadcast them to all clients
+		switch event.Type {
+		case EventTypingStart, EventPresenceUpdate:
+			c.Hub.BroadcastEvent(event)
+		default:
+			c.Hub.logger.Info("received event from client", "type", event.Type, "user_id", c.UserID)
+		}
+	}
+}
+
+// writePump pumps messages from the hub to the websocket connection.
+func (c *Client) writePump() {
+	ticker := time.NewTicker(pingPeriod)
+	defer func() {
+		ticker.Stop()
+		c.Conn.Close()
+	}()
+
+	for {
+		select {
+		case message, ok := <-c.send:
+			c.Conn.SetWriteDeadline(time.Now().Add(writeWait))
+			if !ok {
+				c.Conn.WriteMessage(websocket.CloseMessage, []byte{})
+				return
+			}
+
+			w, err := c.Conn.NextWriter(websocket.TextMessage)
+			if err != nil {
+				return
+			}
+			w.Write(message)
+
+			// Drain queued messages into the current write
+			n := len(c.send)
+			for i := 0; i < n; i++ {
+				w.Write([]byte{'\n'})
+				w.Write(<-c.send)
+			}
+
+			if err := w.Close(); err != nil {
+				return
+			}
+
+		case <-ticker.C:
+			c.Conn.SetWriteDeadline(time.Now().Add(writeWait))
+			if err := c.Conn.WriteMessage(websocket.PingMessage, nil); err != nil {
+				return
+			}
+		}
+	}
+}
+
+// ServeWS handles websocket requests from the peer. It authenticates the
+// client via a session token passed as the "token" query parameter.
+func ServeWS(db *sql.DB, hub *Hub, logger *slog.Logger, w http.ResponseWriter, r *http.Request) {
+	token := r.URL.Query().Get("token")
+	if token == "" {
+		http.Error(w, `{"error":"missing token"}`, http.StatusUnauthorized)
+		return
+	}
+
+	var userID string
+	err := db.QueryRowContext(context.Background(),
+		`SELECT user_id FROM sessions WHERE token = $1 AND expires_at > NOW()`,
+		token,
+	).Scan(&userID)
+	if err != nil {
+		http.Error(w, `{"error":"invalid or expired session"}`, http.StatusUnauthorized)
+		return
+	}
+
+	conn, err := upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		logger.Error("websocket upgrade failed", "error", err)
+		return
+	}
+
+	client := &Client{
+		Hub:    hub,
+		Conn:   conn,
+		UserID: userID,
+		send:   make(chan []byte, 256),
+	}
+
+	hub.Register(client)
+
+	go client.writePump()
+	go client.readPump()
+}
@@ -0,0 +1,50 @@
+package gateway
+
+import "encoding/json"
+
+// Event type constants.
+const (
+	EventMessageCreate  = "MESSAGE_CREATE"
+	EventMessageUpdate  = "MESSAGE_UPDATE"
+	EventMessageDelete  = "MESSAGE_DELETE"
+	EventChannelCreate  = "CHANNEL_CREATE"
+	EventChannelUpdate  = "CHANNEL_UPDATE"
+	EventChannelDelete  = "CHANNEL_DELETE"
+	EventMemberAdd      = "SERVER_MEMBER_ADD"
+	EventMemberRemove   = "SERVER_MEMBER_REMOVE"
+	EventPresenceUpdate = "PRESENCE_UPDATE"
+	EventTypingStart    = "TYPING_START"
+)
+
+// Event represents a WebSocket event sent to clients.
+type Event struct {
+	Type string      `json:"type"`
+	Data interface{} `json:"data,omitempty"`
+}
+
+// MarshalJSON implements custom JSON marshaling for Event.
+func (e Event) MarshalJSON() ([]byte, error) {
+	type eventAlias struct {
+		Type string      `json:"type"`
+		Data interface{} `json:"data,omitempty"`
+	}
+	return json.Marshal(eventAlias{
+		Type: e.Type,
+		Data: e.Data,
+	})
+}
+
+// UnmarshalJSON implements custom JSON unmarshaling for Event.
+func (e *Event) UnmarshalJSON(data []byte) error {
+	type eventAlias struct {
+		Type string          `json:"type"`
+		Data json.RawMessage `json:"data,omitempty"`
+	}
+	var raw eventAlias
+	if err := json.Unmarshal(data, &raw); err != nil {
+		return err
+	}
+	e.Type = raw.Type
+	e.Data = raw.Data
+	return nil
+}
@@ -0,0 +1,90 @@
+package gateway
+
+import (
+	"encoding/json"
+	"log/slog"
+	"sync"
+)
+
+// Hub maintains the set of active clients and broadcasts messages to them.
+type Hub struct {
+	mu         sync.RWMutex
+	clients    map[*Client]struct{}
+	register   chan *Client
+	unregister chan *Client
+	broadcast  chan []byte
+	logger     *slog.Logger
+}
+
+// NewHub creates a new Hub.
+func NewHub(logger *slog.Logger) *Hub {
+	return &Hub{
+		clients:    make(map[*Client]struct{}),
+		register:   make(chan *Client),
+		unregister: make(chan *Client),
+		broadcast:  make(chan []byte, 256),
+		logger:     logger,
+	}
+}
+
+// Run starts the hub's main event loop. Call this in a goroutine.
+func (h *Hub) Run() {
+	for {
+		select {
+		case client := <-h.register:
+			h.mu.Lock()
+			h.clients[client] = struct{}{}
+			h.mu.Unlock()
+			h.logger.Info("client connected", "user_id", client.UserID)
+
+		case client := <-h.unregister:
+			h.mu.Lock()
+			if _, ok := h.clients[client]; ok {
+				delete(h.clients, client)
+				close(client.send)
+			}
+			h.mu.Unlock()
+			h.logger.Info("client disconnected", "user_id", client.UserID)
+
+		case message := <-h.broadcast:
+			h.mu.RLock()
+			for client := range h.clients {
+				select {
+				case client.send <- message:
+				default:
+					go func(c *Client) {
+						h.unregister <- c
+					}(client)
+				}
+			}
+			h.mu.RUnlock()
+		}
+	}
+}
+
+// Register queues a client for registration with the hub.
+func (h *Hub) Register(client *Client) {
+	h.register <- client
+}
+
+// Unregister queues a client for removal from the hub.
+func (h *Hub) Unregister(client *Client) {
+	h.unregister <- client
+}
+
+// BroadcastEvent serializes an Event and broadcasts it to all connected clients.
+func (h *Hub) BroadcastEvent(event Event) {
+	data, err := json.Marshal(event)
+	if err != nil {
+		h.logger.Error("failed to marshal event", "type", event.Type, "error", err)
+		return
+	}
+	h.broadcast <- data
+}
+
+// ClientCount returns the number of connected clients.
+func (h *Hub) ClientCount() int {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	return len(h.clients)
+}