From 998d1b1b50cbb425be731d6595a38968d0318fed Mon Sep 17 00:00:00 2001 From: Bartal Laearsson Date: Fri, 24 Jul 2026 21:42:06 +0100 Subject: [PATCH] add 1 on 1 chat --- cmd/server/chat.go | 216 +++++++++++++++++++++++++-------------------- static/index.css | 57 +++++++++++- static/index.html | 15 +++- static/index.js | 125 ++++++++++++++++++++------ 4 files changed, 285 insertions(+), 128 deletions(-) diff --git a/cmd/server/chat.go b/cmd/server/chat.go index 11d305c..1b0c6d5 100644 --- a/cmd/server/chat.go +++ b/cmd/server/chat.go @@ -2,11 +2,14 @@ package main import ( "context" + "encoding/json" "errors" "io" - "log" + "log/slog" "net" "net/http" + "net/url" + "strings" "sync" "time" @@ -15,68 +18,75 @@ import ( "github.com/coder/websocket" ) -// chatServer enables broadcasting to a set of subscribers. +// chatMessage is the payload sent from the frontend to /publish. +type chatMessage struct { + To string `json:"to"` + Message string `json:"message"` + From string `json:"from,omitempty"` // optional, server fills this in +} + +// chatServer enables 1:1 messaging between named subscribers. type chatServer struct { // subscriberMessageBuffer controls the max number // of messages that can be queued for a subscriber // before it is kicked. - // - // Defaults to 16. subscriberMessageBuffer int // publishLimiter controls the rate limit applied to the publish endpoint. - // - // Defaults to one publish every 100ms with a burst of 8. publishLimiter *rate.Limiter // logf controls where logs are sent. - // Defaults to log.Printf. logf func(f string, v ...any) - // serveMux routes the various endpoints to the appropriate handler. + // serveMux routes the various endpoints. serveMux http.ServeMux - subscribersMu sync.Mutex - subscribers map[*subscriber]struct{} + // subscribersMu protects the subscribers map. + // Key is the username (from query param). + subscribersMu sync.RWMutex + subscribers map[string]*subscriber } // newChatServer constructs a chatServer with the defaults. func newChatServer() *chatServer { cs := &chatServer{ subscriberMessageBuffer: 16, - logf: log.Printf, - subscribers: make(map[*subscriber]struct{}), + logf: slog.Info, + subscribers: make(map[string]*subscriber), publishLimiter: rate.NewLimiter(rate.Every(time.Millisecond*100), 8), } - // Serve static files (index.html, index.js, index.css) from cwd cs.serveMux.Handle("/", http.FileServer(http.Dir("."))) - // WebSocket endpoint: clients connect here to receive messages cs.serveMux.HandleFunc("/subscribe", cs.subscribeHandler) - // HTTP POST endpoint: clients POST here to send a message cs.serveMux.HandleFunc("/publish", cs.publishHandler) + cs.serveMux.HandleFunc("/users", cs.usersHandler) // list online users return cs } // subscriber represents a single connected client. -// Messages are sent on the msgs channel. If the client -// cannot keep up (buffer full), closeSlow is called -// to disconnect them. type subscriber struct { - msgs chan []byte - closeSlow func() + username string // unique identifier from query param + msgs chan []byte // buffered channel for outbound messages + closeSlow func() // kicks the subscriber if buffer fills + cancel context.CancelFunc // cancels the read context on disconnect } -// ServeHTTP makes chatServer implement http.Handler, -// delegating to the internal serveMux. +// ServeHTTP delegates to the internal serveMux. func (cs *chatServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { cs.serveMux.ServeHTTP(w, r) } -// subscribeHandler accepts the WebSocket connection and then subscribes -// it to all future messages. +// subscribeHandler extracts the username from query params and upgrades to WebSocket. func (cs *chatServer) subscribeHandler(w http.ResponseWriter, r *http.Request) { - err := cs.subscribe(w, r) + // Extract username from ?user= parameter + username := r.URL.Query().Get("user") + if username == "" { + http.Error(w, "missing user parameter", http.StatusBadRequest) + return + } + username = url.PathEscape(strings.TrimSpace(username)) + + err := cs.subscribe(w, r, username) if errors.Is(err, context.Canceled) { return } @@ -85,48 +95,20 @@ func (cs *chatServer) subscribeHandler(w http.ResponseWriter, r *http.Request) { return } if err != nil { - cs.logf("%v", err) + cs.logf("subscribe error for user %s: %v", username, err) return } } -// publishHandler reads the request body with a limit of 8192 bytes -// and then publishes the received message. -func (cs *chatServer) publishHandler(w http.ResponseWriter, r *http.Request) { - if r.Method != "POST" { - http.Error(w, http.StatusText(http.StatusMethodNotAllowed), http.StatusMethodNotAllowed) - return - } - body := http.MaxBytesReader(w, r.Body, 8192) - msg, err := io.ReadAll(body) - if err != nil { - http.Error(w, http.StatusText(http.StatusRequestEntityTooLarge), http.StatusRequestEntityTooLarge) - return - } - - cs.publish(msg) - - w.WriteHeader(http.StatusAccepted) -} - -// subscribe upgrades the HTTP connection to a WebSocket, -// registers the subscriber, and then loops forever writing -// messages from the subscriber's channel to the WebSocket. -// -// It uses CloseRead to keep reading control frames (close, ping, pong) -// and cancel the context if the connection drops. This means we -// never actually read message data from the WebSocket — -// messages come in via /publish instead. -func (cs *chatServer) subscribe(w http.ResponseWriter, r *http.Request) error { - // We need a mutex here because the WebSocket connection might - // be closed by closeSlow (from another goroutine) before we've - // assigned it to c. The mutex ensures we don't race. +// subscribe upgrades the HTTP connection to WebSocket and registers the subscriber. +func (cs *chatServer) subscribe(w http.ResponseWriter, r *http.Request, username string) error { var mu sync.Mutex var c *websocket.Conn var closed bool s := &subscriber{ - msgs: make(chan []byte, cs.subscriberMessageBuffer), + username: username, + msgs: make(chan []byte, cs.subscriberMessageBuffer), closeSlow: func() { mu.Lock() defer mu.Unlock() @@ -137,19 +119,17 @@ func (cs *chatServer) subscribe(w http.ResponseWriter, r *http.Request) error { }, } - // Register the subscriber BEFORE accepting the WebSocket. - // This means we won't miss any messages published between - // the accept and the registration. + // Register BEFORE accepting WebSocket cs.addSubscriber(s) defer cs.deleteSubscriber(s) - // Upgrade HTTP to WebSocket + // Upgrade to WebSocket c2, err := websocket.Accept(w, r, nil) if err != nil { return err } - // Check if closeSlow was already called before we got the connection + // Check if already closed mu.Lock() if closed { mu.Unlock() @@ -160,68 +140,116 @@ func (cs *chatServer) subscribe(w http.ResponseWriter, r *http.Request) error { defer c.CloseNow() - // CloseRead returns a context that is cancelled when: - // - The client sends a close frame - // - The connection drops - // This frees us from manually handling ping/pong and deadlines. - ctx := c.CloseRead(context.Background()) + // Create context that cancels on close/read error + ctx, cancel := context.WithCancel(context.Background()) + s.cancel = cancel + + // CloseRead handles ping/pong/close frames automatically + connCtx := c.CloseRead(ctx) + + cs.logf("user %s connected", username) - // Main loop: write messages to the WebSocket as they arrive for { select { case msg := <-s.msgs: - err := writeTimeout(ctx, time.Second*5, c, msg) + err := writeTimeout(connCtx, time.Second*5, c, msg) if err != nil { return err } - case <-ctx.Done(): - // Context cancelled = client disconnected - return ctx.Err() + case <-connCtx.Done(): + cs.logf("user %s disconnected", username) + cancel() // trigger cleanup + return connCtx.Err() } } } -// publish sends a message to ALL subscribers. -// It never blocks — if a subscriber's buffer is full, -// they get kicked via closeSlow. -func (cs *chatServer) publish(msg []byte) { - cs.subscribersMu.Lock() - defer cs.subscribersMu.Unlock() +// publishHandler reads the JSON message and routes it to the recipient. +func (cs *chatServer) publishHandler(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, http.StatusText(http.StatusMethodNotAllowed), http.StatusMethodNotAllowed) + return + } + + body := http.MaxBytesReader(w, r.Body, 8192) + data, err := io.ReadAll(body) + if err != nil { + http.Error(w, http.StatusText(http.StatusRequestEntityTooLarge), http.StatusRequestEntityTooLarge) + return + } + + var msg chatMessage + if err := json.Unmarshal(data, &msg); err != nil { + http.Error(w, "invalid JSON", http.StatusBadRequest) + return + } + + if msg.To == "" || msg.Message == "" { + http.Error(w, "missing 'to' or 'message' field", http.StatusBadRequest) + return + } + + // Deliver the message to the recipient + delivered := cs.deliverMessage(msg.To, msg.Message) + + if delivered { + w.WriteHeader(http.StatusAccepted) + } else { + http.Error(w, "recipient not found", http.StatusNotFound) + } +} + +// deliverMessage sends a message to a specific recipient. +func (cs *chatServer) deliverMessage(toUsername, message string) bool { + cs.subscribersMu.RLock() + recipient, ok := cs.subscribers[toUsername] + cs.subscribersMu.RUnlock() + + if !ok { + return false + } - // Rate limit: wait for a token before broadcasting cs.publishLimiter.Wait(context.Background()) - for s := range cs.subscribers { - select { - case s.msgs <- msg: - // Message delivered to subscriber's buffer - default: - // Buffer is full — kick the slow subscriber - go s.closeSlow() - } + select { + case recipient.msgs <- []byte(message): + return true + default: + // Buffer full — kick slow client + go recipient.closeSlow() + return false } } +// usersHandler returns list of currently connected usernames. +func (cs *chatServer) usersHandler(w http.ResponseWriter, r *http.Request) { + cs.subscribersMu.RLock() + users := make([]string, 0, len(cs.subscribers)) + for username := range cs.subscribers { + users = append(users, username) + } + cs.subscribersMu.RUnlock() + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(users) +} + // addSubscriber registers a subscriber. func (cs *chatServer) addSubscriber(s *subscriber) { cs.subscribersMu.Lock() - cs.subscribers[s] = struct{}{} + cs.subscribers[s.username] = s cs.subscribersMu.Unlock() } // deleteSubscriber removes a subscriber. func (cs *chatServer) deleteSubscriber(s *subscriber) { cs.subscribersMu.Lock() - delete(cs.subscribers, s) + delete(cs.subscribers, s.username) cs.subscribersMu.Unlock() } -// writeTimeout writes a message to the WebSocket with a deadline. -// If the write doesn't complete within the timeout, the context -// is cancelled and the operation aborts. func writeTimeout(ctx context.Context, timeout time.Duration, c *websocket.Conn, msg []byte) error { ctx, cancel := context.WithTimeout(ctx, timeout) defer cancel() - return c.Write(ctx, websocket.MessageText, msg) } diff --git a/static/index.css b/static/index.css index 734cd1e..d9fd281 100644 --- a/static/index.css +++ b/static/index.css @@ -13,13 +13,52 @@ body { color: #e0e0e0; } -#messages { +.header { + display: flex; + justify-content: space-between; + align-items: center; + padding: 0.5rem 1rem; + background: #0f0f23; + border-bottom: 1px solid #333; +} + +.user-setup { + display: flex; + gap: 0.5rem; +} + +#username { + padding: 0.4rem; + background: #16213e; + border: 1px solid #333; + border-radius: 4px; + color: #e0e0e0; +} + +#login-btn { + padding: 0.4rem 0.8rem; + background: #6d4aff; + color: white; + border: none; + border-radius: 4px; + cursor: pointer; +} + +.online-users { + font-size: 0.9rem; +} + +#online-list { + color: #4ade80; +} + +#message-log { flex: 1; overflow-y: auto; padding: 1rem; } -.message { +#message-log p { padding: 0.4rem 0.6rem; margin-bottom: 0.3rem; background: #16213e; @@ -32,9 +71,10 @@ body { padding: 1rem; background: #0f0f23; border-top: 1px solid #333; + gap: 0.5rem; } -#message-input { +#recipient-select { flex: 1; padding: 0.5rem; background: #16213e; @@ -44,8 +84,17 @@ body { font-family: monospace; } +#message-input { + flex: 2; + padding: 0.5rem; + background: #16213e; + border: 1px solid #333; + border-radius: 4px; + color: #e0e0e0; + font-family: monospace; +} + #publish-form button { - margin-left: 0.5rem; padding: 0.5rem 1rem; background: #6d4aff; color: white; diff --git a/static/index.html b/static/index.html index 76229e6..3740b2a 100644 --- a/static/index.html +++ b/static/index.html @@ -7,10 +7,21 @@ +
+
+ + +
+
+ Online: +
+
+
-
- + + +
diff --git a/static/index.js b/static/index.js index de33fa0..f0c2d25 100644 --- a/static/index.js +++ b/static/index.js @@ -1,22 +1,54 @@ (() => { + let ws = null let expectingMessage = false + const currentUserEl = document.getElementById('username') + const loginBtn = document.getElementById('login-btn') + const messageLog = document.getElementById('message-log') + const publishForm = document.getElementById('publish-form') + const recipientSelect = document.getElementById('recipient-select') + const messageInput = document.getElementById('message-input') + const onlineList = document.getElementById('online-list') - function dial() { - const conn = new WebSocket(`ws://${location.host}/ws/subscribe`) + // Auto-fill username from URL parameter + const urlParams = new URLSearchParams(window.location.search) + const urlUser = urlParams.get('user') + if (urlUser) { + currentUserEl.value = urlUser + } - conn.addEventListener('close', ev => { - appendLog(`WebSocket Disconnected code: ${ev.code}, reason: ${ev.reason}`, true) - if (ev.code !== 1001) { - appendLog('Reconnecting in 1s', true) - setTimeout(dial, 1000) - } - }) + // Login button handler + loginBtn.addEventListener('click', () => { + const username = currentUserEl.value.trim() + if (!username) { + appendLog('Please enter a username', true) + return + } - conn.addEventListener('open', ev => { + // Connect to WebSocket with username + connect(username) + + // Show publish form, hide login + loginBtn.style.display = 'none' + currentUserEl.style.display = 'none' + publishForm.style.display = 'flex' + messageInput.disabled = false + messageInput.focus() + + appendLog(`Joined as ${username}`) + }) + + function connect(username) { + const encodedUser = encodeURIComponent(username) + ws = new WebSocket(`ws://${location.host}/ws/subscribe?user=${encodedUser}`) + + ws.addEventListener('open', () => { console.info('WebSocket connected') + refreshUsers() + // Refresh users every 5 seconds + setInterval(refreshUsers, 5000) }) - conn.addEventListener('message', ev => { + ws.addEventListener('message', (ev) => { if (typeof ev.data !== 'string') { console.error('unexpected message type', typeof ev.data) return @@ -27,47 +59,84 @@ expectingMessage = false } }) + + ws.addEventListener('close', (ev) => { + appendLog(`Disconnected (code: ${ev.code})`, true) + if (ev.code !== 1001) { + appendLog('Reconnecting...', true) + setTimeout(() => connect(username), 2000) + } + }) + + ws.addEventListener('error', (err) => { + console.error('WebSocket error:', err) + }) } - dial() + function refreshUsers() { + fetch('/ws/users') + .then(r => r.json()) + .then(users => { + onlineList.textContent = users.join(', ') || 'none' - const messageLog = document.getElementById('message-log') - const publishForm = document.getElementById('publish-form') - const messageInput = document.getElementById('message-input') + // Update recipient dropdown + recipientSelect.innerHTML = '' + if (users.length > 0) { + users.forEach(user => { + const opt = document.createElement('option') + opt.value = user + opt.textContent = user + recipientSelect.appendChild(opt) + }) + } + }) + .catch(err => { + console.error('Failed to refresh users:', err) + }) + } - function appendLog(text, error) { + function appendLog(text, error = false) { const p = document.createElement('p') - p.innerText = `${new Date().toLocaleTimeString()}: ${text}` + const time = new Date().toLocaleTimeString() + p.innerText = `[${time}] ${text}` if (error) { - p.style.color = 'red' - p.style.fontStyle = 'bold' + p.style.color = '#ff6b6b' } messageLog.append(p) return p } - appendLog('Submit a message to get started!') - - publishForm.onsubmit = async ev => { + // Submit message + publishForm.onsubmit = async (ev) => { ev.preventDefault() - const msg = messageInput.value - if (msg === '') { + const recipient = recipientSelect.value.trim() + const msg = messageInput.value.trim() + + if (!recipient || !msg) { + appendLog('Please select a recipient and enter a message', true) return } - messageInput.value = '' + messageInput.value = '' expectingMessage = true + try { + const payload = JSON.stringify({ to: recipient, message: msg }) const resp = await fetch('/ws/publish', { method: 'POST', - body: msg, + headers: { 'Content-Type': 'application/json' }, + body: payload, }) + if (resp.status !== 202) { - throw new Error(`Unexpected HTTP Status ${resp.status} ${resp.statusText}`) + const text = await resp.text() + throw new Error(`Failed: ${text}`) } } catch (err) { - appendLog(`Publish failed: ${err.message}`, true) + appendLog(`Send failed: ${err.message}`, true) } } + + appendLog('Enter your username to join the chat') })()