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 @@
+
+
-
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')
})()