Design a chat system

package main
 
import "fmt"
 
type message struct {
	seq  int
	text string
}
 
// room keeps an ordered history; since(n) is what a reconnecting client asks for.
type room struct{ history []message }
 
func (r *room) post(text string) message {
	m := message{seq: len(r.history) + 1, text: text}
	r.history = append(r.history, m)
	return m
}
 
func (r *room) since(seq int) []message { return r.history[seq:] }
 
func main() {
	r := &room{}
	lastSeen := 0
	for _, t := range []string{"hi", "anyone here?", "yes!"} {
		lastSeen = r.post(t).seq // the phone is online and receives these
	}
	fmt.Println("phone saw up to seq", lastSeen, "then lost signal")
 
	r.post("did you see the game?")
	r.post("unbelievable ending")
 
	for _, m := range r.since(lastSeen) {
		fmt.Printf("replayed #%d %q\n", m.seq, m.text)
	}
}

phone saw up to seq 3 then lost signal
replayed #4 "did you see the game?"
replayed #5 "unbelievable ending"

chat/
├── go.mod
├── main.go                     HTTP API: post, SSE stream, online list
├── main_test.go                end-to-end SSE tests
├── internal/chat/hub.go        rooms, sequence numbers, fan-out
├── internal/chat/hub_test.go
└── web/index.html              the chat client

mkdir chat && cd chat && go mod init chat
mkdir -p internal/chat web

module chat
 
go 1.24

// Package chat implements rooms with ordered, replayable history and live subscribers.
package chat
 
import (
	"errors"
	"slices"
	"sync"
	"time"
	"unicode/utf8"
)
 
// MaxText is the longest message accepted, in characters.
const MaxText = 2000
 
// Message is one chat message. Seq is its position in the room, starting at 1.
type Message struct {
	Seq      int64     `json:"seq"`
	Room     string    `json:"room"`
	From     string    `json:"from"`
	Text     string    `json:"text"`
	At       time.Time `json:"at"`
	ClientID string    `json:"client_id,omitempty"`
}
 
var ErrInvalid = errors.New("message must have a sender and 1 to 2000 characters of text")
 
type room struct {
	seq      int64
	history  []Message          // oldest first, at most historyLimit entries
	byClient map[string]Message // client ID -> message, for messages still in history
	subs     map[*Sub]struct{}
}
 
// Sub is a live subscription. Messages arrive on C; C is closed when the subscription
// ends, either by Close or because the subscriber fell too far behind.
type Sub struct {
	C      <-chan Message
	c      chan Message
	room   *room
	user   string
	hub    *Hub
	closed bool
}
 
// Hub holds every room.
type Hub struct {
	mu           sync.Mutex
	rooms        map[string]*room
	historyLimit int
	bufSize      int
	now          func() time.Time
}
 
// NewHub keeps the last historyLimit messages per room and buffers bufSize messages per subscriber.
func NewHub(historyLimit, bufSize int, now func() time.Time) *Hub {
	if now == nil {
		now = time.Now
	}
	return &Hub{rooms: map[string]*room{}, historyLimit: historyLimit, bufSize: bufSize, now: now}
}
 
func (h *Hub) roomLocked(name string) *room {
	r, ok := h.rooms[name]
	if !ok {
		r = &room{byClient: map[string]Message{}, subs: map[*Sub]struct{}{}}
		h.rooms[name] = r
	}
	return r
}
 
// Post stores a message with the room's next sequence number and delivers it to subscribers.
// Resending with the same clientID returns the original message instead of storing a duplicate.
func (h *Hub) Post(roomName, from, text, clientID string) (Message, error) {
	if from == "" || text == "" || utf8.RuneCountInString(text) > MaxText {
		return Message{}, ErrInvalid
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	r := h.roomLocked(roomName)
	if m, ok := r.byClient[clientID]; ok && clientID != "" {
		return m, nil
	}
	r.seq++
	m := Message{Seq: r.seq, Room: roomName, From: from, Text: text, At: h.now().UTC(), ClientID: clientID}
	r.history = append(r.history, m)
	if clientID != "" {
		r.byClient[clientID] = m
	}
	if len(r.history) > h.historyLimit {
		old := r.history[0]
		r.history = slices.Delete(r.history, 0, 1)
		delete(r.byClient, old.ClientID)
	}
	for s := range r.subs {
		select {
		case s.c <- m:
		default:
			// Too slow: disconnect rather than buffer without limit. The client reconnects
			// with its last seq and catches up from history.
			s.closeLocked()
		}
	}
	return m, nil
}
 
// Subscribe returns the stored messages after seq since (the catch-up) and a live
// subscription for everything newer. Holding the lock across both means no message
// can fall between the backlog and the subscription.
func (h *Hub) Subscribe(roomName, user string, since int64) ([]Message, *Sub) {
	h.mu.Lock()
	defer h.mu.Unlock()
	r := h.roomLocked(roomName)
	i, _ := slices.BinarySearchFunc(r.history, since+1, func(m Message, seq int64) int { return int(m.Seq - seq) })
	backlog := slices.Clone(r.history[i:])
	c := make(chan Message, h.bufSize)
	s := &Sub{C: c, c: c, room: r, user: user, hub: h}
	r.subs[s] = struct{}{}
	return backlog, s
}
 
// Close ends the subscription. It is safe to call more than once.
func (s *Sub) Close() {
	s.hub.mu.Lock()
	defer s.hub.mu.Unlock()
	s.closeLocked()
}
 
func (s *Sub) closeLocked() {
	if s.closed {
		return
	}
	s.closed = true
	delete(s.room.subs, s)
	close(s.c)
}
 
// Online lists the distinct users subscribed to a room, sorted.
func (h *Hub) Online(roomName string) []string {
	h.mu.Lock()
	defer h.mu.Unlock()
	users := []string{}
	if r, ok := h.rooms[roomName]; ok {
		for s := range r.subs {
			users = append(users, s.user)
		}
	}
	slices.Sort(users)
	return slices.Compact(users)
}

package chat
 
import (
	"errors"
	"slices"
	"strings"
	"testing"
	"time"
)
 
func newHub(limit, buf int) *Hub {
	t0 := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
	return NewHub(limit, buf, func() time.Time { return t0 })
}
 
func texts(ms []Message) []string {
	var out []string
	for _, m := range ms {
		out = append(out, m.Text)
	}
	return out
}
 
func TestSequenceAndCatchUp(t *testing.T) {
	h := newHub(100, 10)
	for _, s := range []string{"a", "b", "c", "d"} {
		h.Post("go", "ann", s, "")
	}
	m, _ := h.Post("rust", "ann", "other room", "")
	if m.Seq != 1 {
		t.Errorf("each room counts from 1, got %d", m.Seq)
	}
	backlog, sub := h.Subscribe("go", "bob", 2)
	defer sub.Close()
	if got := texts(backlog); !slices.Equal(got, []string{"c", "d"}) {
		t.Fatalf("catch-up after 2 = %v", got)
	}
	h.Post("go", "ann", "e", "")
	if live := <-sub.C; live.Seq != 5 || live.Text != "e" {
		t.Errorf("live message = %+v", live)
	}
}
 
func TestHistoryLimit(t *testing.T) {
	h := newHub(3, 10)
	for _, s := range []string{"1", "2", "3", "4", "5"} {
		h.Post("r", "ann", s, "")
	}
	backlog, sub := h.Subscribe("r", "bob", 0)
	sub.Close()
	if got := texts(backlog); !slices.Equal(got, []string{"3", "4", "5"}) {
		t.Errorf("history = %v, want the last 3", got)
	}
}
 
func TestResendIsDeduplicated(t *testing.T) {
	h := newHub(100, 10)
	first, _ := h.Post("r", "ann", "hello", "c-1")
	again, _ := h.Post("r", "ann", "hello", "c-1")
	if again.Seq != first.Seq {
		t.Errorf("resend got seq %d, want the original %d", again.Seq, first.Seq)
	}
	backlog, sub := h.Subscribe("r", "bob", 0)
	sub.Close()
	if len(backlog) != 1 {
		t.Errorf("stored %d messages, want 1", len(backlog))
	}
}
 
func TestSlowSubscriberIsDropped(t *testing.T) {
	h := newHub(100, 1)
	_, slow := h.Subscribe("r", "slow", 0)
	_, fast := h.Subscribe("r", "fast", 0)
	h.Post("r", "ann", "one", "")
	<-fast.C                      // fast keeps up
	h.Post("r", "ann", "two", "") // slow's buffer of 1 is full: it gets disconnected
	<-fast.C
	if m := <-slow.C; m.Text != "one" {
		t.Errorf("buffered message = %q", m.Text)
	}
	if _, ok := <-slow.C; ok {
		t.Error("slow subscriber should be closed")
	}
	if got := h.Online("r"); !slices.Equal(got, []string{"fast"}) {
		t.Errorf("online = %v", got)
	}
	slow.Close() // already closed by the hub: must not panic
	fast.Close()
}
 
func TestOnlineAndValidation(t *testing.T) {
	h := newHub(100, 10)
	_, a1 := h.Subscribe("r", "ann", 0)
	_, a2 := h.Subscribe("r", "ann", 0) // two tabs
	_, b := h.Subscribe("r", "bob", 0)
	if got := h.Online("r"); !slices.Equal(got, []string{"ann", "bob"}) {
		t.Errorf("online = %v", got)
	}
	a1.Close()
	a2.Close()
	b.Close()
	if got := h.Online("r"); len(got) != 0 {
		t.Errorf("online after close = %v", got)
	}
	for _, bad := range []string{"", strings.Repeat("x", MaxText+1)} {
		if _, err := h.Post("r", "ann", bad, ""); !errors.Is(err, ErrInvalid) {
			t.Errorf("text of %d chars: err = %v", len(bad), err)
		}
	}
}

package main
 
import (
	_ "embed"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"strconv"
	"time"
 
	"chat/internal/chat"
)
 
//go:embed web/index.html
var indexHTML []byte
 
func writeJSON(w http.ResponseWriter, code int, v any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(code)
	json.NewEncoder(w).Encode(v)
}
 
func newServer(hub *chat.Hub, heartbeat time.Duration) http.Handler {
	mux := http.NewServeMux()
	mux.HandleFunc("GET /{$}", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/html; charset=utf-8")
		w.Write(indexHTML)
	})
 
	mux.HandleFunc("POST /api/rooms/{room}/messages", func(w http.ResponseWriter, r *http.Request) {
		var in struct {
			From     string `json:"from"`
			Text     string `json:"text"`
			ClientID string `json:"client_id"`
		}
		if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<14)).Decode(&in); err != nil {
			writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON"})
			return
		}
		m, err := hub.Post(r.PathValue("room"), in.From, in.Text, in.ClientID)
		if errors.Is(err, chat.ErrInvalid) {
			writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
			return
		}
		writeJSON(w, http.StatusCreated, m)
	})
 
	mux.HandleFunc("GET /api/rooms/{room}/online", func(w http.ResponseWriter, r *http.Request) {
		writeJSON(w, http.StatusOK, hub.Online(r.PathValue("room")))
	})
 
	mux.HandleFunc("GET /api/rooms/{room}/events", func(w http.ResponseWriter, r *http.Request) {
		// The browser sends Last-Event-ID on automatic reconnects; ?since= covers the first connect.
		since, _ := strconv.ParseInt(r.Header.Get("Last-Event-ID"), 10, 64)
		if since == 0 {
			since, _ = strconv.ParseInt(r.URL.Query().Get("since"), 10, 64)
		}
		user := r.URL.Query().Get("user")
		if user == "" {
			user = "anonymous"
		}
		backlog, sub := hub.Subscribe(r.PathValue("room"), user, since)
		defer sub.Close()
 
		h := w.Header()
		h.Set("Content-Type", "text/event-stream")
		h.Set("Cache-Control", "no-cache")
		h.Set("X-Accel-Buffering", "no") // tell nginx-style proxies not to buffer the stream
		rc := http.NewResponseController(w)
 
		send := func(m chat.Message) error {
			b, _ := json.Marshal(m)
			if _, err := fmt.Fprintf(w, "id: %d\nevent: message\ndata: %s\n\n", m.Seq, b); err != nil {
				return err
			}
			return rc.Flush()
		}
		fmt.Fprint(w, "retry: 1000\n\n") // reconnect after 1s if the stream breaks
		for _, m := range backlog {
			if send(m) != nil {
				return
			}
		}
		rc.Flush()
 
		ping := time.NewTicker(heartbeat) // keeps idle connections open through proxies
		defer ping.Stop()
		for {
			select {
			case <-r.Context().Done():
				return
			case m, ok := <-sub.C:
				if !ok {
					return // dropped as a slow consumer: the client reconnects and catches up
				}
				if send(m) != nil {
					return
				}
			case <-ping.C:
				fmt.Fprint(w, ": ping\n\n")
				if rc.Flush() != nil {
					return
				}
			}
		}
	})
	return mux
}
 
func main() {
	hub := chat.NewHub(500, 64, nil)
	port := os.Getenv("PORT")
	if port == "" {
		port = "8080"
	}
	slog.Info("listening", "port", port)
	// No WriteTimeout: SSE responses are meant to stay open.
	srv := &http.Server{Addr: ":" + port, Handler: newServer(hub, 25*time.Second), ReadHeaderTimeout: 5 * time.Second}
	if err := srv.ListenAndServe(); err != nil {
		slog.Error("server", "err", err)
		os.Exit(1)
	}
}

package main
 
import (
	"bufio"
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"slices"
	"strings"
	"testing"
	"time"
 
	"chat/internal/chat"
)
 
// readEvents returns the data of the next n "message" events on the stream.
func readEvents(t *testing.T, sc *bufio.Scanner, n int) []chat.Message {
	t.Helper()
	var out []chat.Message
	for len(out) < n && sc.Scan() {
		if data, ok := strings.CutPrefix(sc.Text(), "data: "); ok {
			var m chat.Message
			if err := json.Unmarshal([]byte(data), &m); err != nil {
				t.Fatal(err)
			}
			out = append(out, m)
		}
	}
	if len(out) < n {
		t.Fatalf("stream ended after %d of %d events: %v", len(out), n, sc.Err())
	}
	return out
}
 
func openStream(t *testing.T, url, lastID string) *bufio.Scanner {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	t.Cleanup(cancel)
	req, _ := http.NewRequestWithContext(ctx, "GET", url, nil)
	if lastID != "" {
		req.Header.Set("Last-Event-ID", lastID)
	}
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { res.Body.Close() })
	if ct := res.Header.Get("Content-Type"); ct != "text/event-stream" {
		t.Fatalf("Content-Type = %q", ct)
	}
	return bufio.NewScanner(res.Body)
}
 
func post(t *testing.T, base, room, body string) *http.Response {
	t.Helper()
	res, err := http.Post(base+"/api/rooms/"+room+"/messages", "application/json", strings.NewReader(body))
	if err != nil {
		t.Fatal(err)
	}
	res.Body.Close()
	return res
}
 
func TestLiveStream(t *testing.T) {
	srv := httptest.NewServer(newServer(chat.NewHub(100, 16, nil), time.Minute))
	t.Cleanup(srv.Close) // runs after the stream cleanups, so no connection is left open
 
	sc := openStream(t, srv.URL+"/api/rooms/go/events?user=bob", "")
	if res := post(t, srv.URL, "go", `{"from":"ann","text":"hello bob","client_id":"c1"}`); res.StatusCode != 201 {
		t.Fatalf("post: %d", res.StatusCode)
	}
	got := readEvents(t, sc, 1)[0]
	if got.Seq != 1 || got.From != "ann" || got.Text != "hello bob" {
		t.Errorf("event = %+v", got)
	}
 
	res, _ := http.Get(srv.URL + "/api/rooms/go/online")
	var online []string
	json.NewDecoder(res.Body).Decode(&online)
	res.Body.Close()
	if !slices.Equal(online, []string{"bob"}) {
		t.Errorf("online = %v", online)
	}
}
 
func TestResumeWithLastEventID(t *testing.T) {
	srv := httptest.NewServer(newServer(chat.NewHub(100, 16, nil), time.Minute))
	t.Cleanup(srv.Close)
	for _, text := range []string{"one", "two", "three"} {
		post(t, srv.URL, "go", `{"from":"ann","text":"`+text+`"}`)
	}
	sc := openStream(t, srv.URL+"/api/rooms/go/events", "1")
	var got []string
	for _, m := range readEvents(t, sc, 2) {
		got = append(got, m.Text)
	}
	if !slices.Equal(got, []string{"two", "three"}) {
		t.Errorf("replayed %v, want [two three]", got)
	}
}
 
func TestRejectsBadMessages(t *testing.T) {
	srv := httptest.NewServer(newServer(chat.NewHub(100, 16, nil), time.Minute))
	t.Cleanup(srv.Close)
	for _, body := range []string{`{"from":"ann","text":""}`, `{"text":"no sender"}`, `nope`} {
		if res := post(t, srv.URL, "go", body); res.StatusCode != 400 {
			t.Errorf("%s: %d, want 400", body, res.StatusCode)
		}
	}
}

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Chat</title>
  <style>
    body { font-family: system-ui, sans-serif; max-width: 760px; margin: 2rem auto; padding: 0 1rem; }
    input, button { padding: .5rem .7rem; border-radius: 8px; border: 1px solid #999; font: inherit; }
    button { background: #fff; cursor: pointer; }
    #wrap { display: grid; grid-template-columns: 1fr 160px; gap: 1rem; margin-top: 1rem; }
    #log { height: 420px; overflow-y: auto; border: 1px solid #ddd; border-radius: 10px; padding: .75rem; }
    .m { margin: .3rem 0; } .m b { margin-right: .4rem; } .m small { color: #888; margin-right: .4rem; font-family: monospace; }
    #online { border: 1px solid #ddd; border-radius: 10px; padding: .75rem; font-size: 14px; }
    #state { font-size: 13px; color: #888; }
    form { display: flex; gap: .5rem; margin-top: .75rem; } form input { flex: 1; }
  </style>
</head>
<body>
  <h1>Chat</h1>
  <p>
    Room <input id="room" value="general" size="10">
    Name <input id="nick" size="10">
    <button id="join">Join</button>
    <span id="state">not connected</span>
  </p>
  <div id="wrap">
    <div>
      <div id="log"></div>
      <form id="send"><input id="text" placeholder="Message" autocomplete="off" disabled><button disabled>Send</button></form>
    </div>
    <div id="online"><b>Online</b><div id="who"></div></div>
  </div>
  <script>
    nick.value = "user" + Math.floor(Math.random() * 1000);
    let es = null, lastSeq = 0, timer = null;
    function show(m) {
      if (m.seq <= lastSeq) return; // already shown (replay overlap)
      lastSeq = m.seq;
      const d = document.createElement("div");
      d.className = "m";
      const seq = document.createElement("small"); seq.textContent = "#" + m.seq;
      const who = document.createElement("b"); who.textContent = m.from;
      d.append(seq, who, document.createTextNode(m.text));
      log.append(d);
      log.scrollTop = log.scrollHeight;
    }
    async function refreshOnline() {
      const users = await (await fetch(`/api/rooms/${encodeURIComponent(room.value)}/online`)).json();
      who.textContent = users.join(", ") || "nobody";
    }
    document.getElementById("join").onclick = () => {
      if (es) es.close();
      clearInterval(timer);
      log.innerHTML = ""; lastSeq = 0;
      es = new EventSource(`/api/rooms/${encodeURIComponent(room.value)}/events?user=${encodeURIComponent(nick.value)}`);
      es.addEventListener("message", (e) => show(JSON.parse(e.data)));
      es.onopen = () => { state.textContent = "connected"; refreshOnline(); };
      es.onerror = () => { state.textContent = "reconnecting…"; };
      text.disabled = false; send.querySelector("button").disabled = false; text.focus();
      timer = setInterval(refreshOnline, 3000);
    };
    send.onsubmit = async (e) => {
      e.preventDefault();
      if (!text.value.trim()) return;
      const body = { from: nick.value, text: text.value, client_id: crypto.randomUUID() };
      text.value = "";
      // A real client retries this POST with the same client_id until it gets a 201.
      await fetch(`/api/rooms/${encodeURIComponent(room.value)}/messages`, { method: "POST", body: JSON.stringify(body) });
    };
  </script>
</body>
</html>

go test -race ./...
go run .
# open http://localhost:8080 in two windows, join the same room and talk
# watch the raw stream from a terminal:
curl -N "localhost:8080/api/rooms/general/events?user=terminal"
# and post from another one:
curl -X POST localhost:8080/api/rooms/general/messages -d '{"from":"curl","text":"hi from the terminal"}'