backend barebone done, big chunk of work done

server/internal/realtime/client.go

@@ -0,0 +1,136 @@
+package realtime
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	"github.com/coder/websocket"
+)
+
+const sendBufferSize = 64
+
+// client wraps a single WebSocket connection belonging to a user.
+type client struct {
+	hub    *Hub
+	conn   *websocket.Conn
+	userID int64
+	send   chan []byte
+	done   chan struct{}
+}
+
+func newClient(hub *Hub, conn *websocket.Conn, userID int64) *client {
+	return &client{
+		hub:    hub,
+		conn:   conn,
+		userID: userID,
+		send:   make(chan []byte, sendBufferSize),
+		done:   make(chan struct{}),
+	}
+}
+
+// enqueue marshals env and queues it for delivery; drops the message if the
+// client cannot keep up.
+func (c *client) enqueue(env envelope) {
+	payload, err := json.Marshal(env)
+	if err != nil {
+		return
+	}
+	c.enqueueRaw(payload)
+}
+
+func (c *client) enqueueRaw(payload []byte) {
+	select {
+	case <-c.done:
+	case c.send <- payload:
+	default:
+		c.closeSlow()
+	}
+}
+
+// closeSlow terminates a connection that cannot keep up with its send queue.
+func (c *client) closeSlow() {
+	c.hub.mu.Lock()
+	defer c.hub.mu.Unlock()
+	if conns := c.hub.conns[c.userID]; conns != nil {
+		if _, ok := conns[c]; ok {
+			c.conn.Close(websocket.StatusPolicyViolation, "client too slow")
+		}
+	}
+}
+
+// writeLoop drains the send queue onto the connection.
+func (c *client) writeLoop() {
+	ctx := context.Background()
+	for {
+		select {
+		case <-c.done:
+			return
+		case msg := <-c.send:
+			writeCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
+			err := c.conn.Write(writeCtx, websocket.MessageText, msg)
+			cancel()
+			if err != nil {
+				return
+			}
+		}
+	}
+}
+
+// readLoop processes inbound client events until the connection closes.
+func (c *client) readLoop() error {
+	ctx := context.Background()
+	for {
+		_, data, err := c.conn.Read(ctx)
+		if err != nil {
+			return err
+		}
+		var env envelope
+		if err := json.Unmarshal(data, &env); err != nil {
+			c.conn.Close(websocket.StatusInvalidFramePayloadData, "bad json")
+			return err
+		}
+
+		switch env.Type {
+		case EventTyping:
+			var msg clientTyping
+			if err := json.Unmarshal(env.Data, &msg); err != nil {
+				continue
+			}
+			if err := c.handleTyping(ctx, msg); err != nil {
+				return err
+			}
+		case EventRead:
+			var msg clientRead
+			if err := json.Unmarshal(env.Data, &msg); err != nil {
+				continue
+			}
+			if err := c.handleRead(ctx, msg); err != nil {
+				return err
+			}
+		default:
+			return websocket.CloseError{Code: websocket.StatusUnsupportedData, Reason: "unsupported event type"}
+		}
+	}
+}
+
+func (c *client) handleTyping(ctx context.Context, msg clientTyping) error {
+	ok, err := c.hub.members.IsMember(ctx, msg.ConversationID, c.userID)
+	if err != nil || !ok {
+		return nil
+	}
+	c.hub.PublishTyping(msg.ConversationID, c.userID, msg.IsTyping)
+	return nil
+}
+
+func (c *client) handleRead(ctx context.Context, msg clientRead) error {
+	ok, err := c.hub.members.IsMember(ctx, msg.ConversationID, c.userID)
+	if err != nil || !ok {
+		return nil
+	}
+	if err := c.hub.members.UpdateLastRead(ctx, msg.ConversationID, c.userID); err != nil {
+		return err
+	}
+	c.hub.PublishRead(msg.ConversationID, c.userID, msg.MessageID)
+	return nil
+}

server/internal/realtime/events.go

@@ -0,0 +1,70 @@
+// Package realtime implements the WebSocket hub: real-time delivery of
+// conversation events (messages, typing, presence, read receipts).
+package realtime
+
+import (
+	"encoding/json"
+	"time"
+)
+
+// Event type constants sent over the wire.
+const (
+	EventMessageCreated = "message.created"
+	EventTyping         = "typing"
+	EventPresence       = "presence"
+	EventRead           = "read"
+)
+
+// Presence status values.
+const (
+	StatusOnline  = "online"
+	StatusOffline = "offline"
+)
+
+// MessageCreated is published when a message is stored.
+type MessageCreated struct {
+	TempID         string    `json:"temp_id,omitempty"`
+	ConversationID int64     `json:"conversation_id"`
+	MessageID      int64     `json:"message_id"`
+	SenderID       int64     `json:"sender_id"`
+	Body           string    `json:"body"`
+	CreatedAt      time.Time `json:"created_at"`
+}
+
+// Typing is published when a user starts or stops typing in a conversation.
+type Typing struct {
+	ConversationID int64 `json:"conversation_id"`
+	UserID         int64 `json:"user_id"`
+	IsTyping       bool  `json:"is_typing"`
+}
+
+// Presence is published when a user comes online or goes offline.
+type Presence struct {
+	UserID int64  `json:"user_id"`
+	Status string `json:"status"`
+}
+
+// Read is published when a user marks messages in a conversation as read.
+type Read struct {
+	ConversationID int64 `json:"conversation_id"`
+	UserID         int64 `json:"user_id"`
+	MessageID      int64 `json:"message_id"`
+}
+
+// envelope is the wire format for every message, both inbound and outbound.
+type envelope struct {
+	Type string          `json:"type"`
+	Data json.RawMessage `json:"data"`
+}
+
+// clientTyping is sent by the client to signal typing state.
+type clientTyping struct {
+	ConversationID int64 `json:"conversation_id"`
+	IsTyping       bool  `json:"is_typing"`
+}
+
+// clientRead is sent by the client to mark messages as read.
+type clientRead struct {
+	ConversationID int64 `json:"conversation_id"`
+	MessageID      int64 `json:"message_id"`
+}

server/internal/realtime/hub.go

@@ -0,0 +1,163 @@
+package realtime
+
+import (
+	"context"
+	"encoding/json"
+	"log/slog"
+	"net/http"
+	"sync"
+
+	"github.com/coder/websocket"
+)
+
+// MembershipProvider resolves conversation membership and read state.
+type MembershipProvider interface {
+	ConversationIDsForUser(ctx context.Context, userID int64) ([]int64, error)
+	IsMember(ctx context.Context, conversationID, userID int64) (bool, error)
+	UpdateLastRead(ctx context.Context, conversationID, userID int64) error
+}
+
+// Hub tracks WebSocket connections and fans out events to the
+// users subscribed to a conversation.
+type Hub struct {
+	mu      sync.Mutex
+	conns   map[int64]map[*client]struct{}
+	subs    map[int64]map[int64]struct{}
+	members MembershipProvider
+}
+
+// NewHub creates a hub that resolves membership through members.
+func NewHub(members MembershipProvider) *Hub {
+	return &Hub{
+		conns:   make(map[int64]map[*client]struct{}),
+		subs:    make(map[int64]map[int64]struct{}),
+		members: members,
+	}
+}
+
+// HandleWS upgrades the HTTP request to a WebSocket connection and
+// runs the client loops until the connection closes.
+func (h *Hub) HandleWS(w http.ResponseWriter, r *http.Request, userID int64) {
+	conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{OriginPatterns: []string{"*"}})
+	if err != nil {
+		slog.Error("ws upgrade failed", "error", err)
+		return
+	}
+	conn.SetReadLimit(32 << 10)
+
+	ctx := r.Context()
+	conversationIDs, err := h.members.ConversationIDsForUser(ctx, userID)
+	if err != nil {
+		slog.Error("ws: load memberships", "user_id", userID, "error", err)
+		conn.Close(websocket.StatusInternalError, "membership lookup failed")
+		return
+	}
+
+	c := newClient(h, conn, userID)
+	h.register(c, conversationIDs)
+	defer h.unregister(c, conversationIDs)
+
+	h.broadcastPresence(userID, conversationIDs, StatusOnline)
+	defer h.broadcastPresence(userID, conversationIDs, StatusOffline)
+
+	go c.writeLoop()
+	if err := c.readLoop(); err != nil {
+		slog.Debug("ws connection closed", "user_id", userID, "error", err)
+	}
+}
+
+// PublishMessageCreated delivers a stored message to everyone subscribed
+// to the conversation, including the sender.
+func (h *Hub) PublishMessageCreated(conversationID int64, evt MessageCreated) {
+	h.sendToConversation(conversationID, envelope{Type: EventMessageCreated, Data: mustMarshal(evt)}, 0)
+}
+
+// PublishTyping delivers a typing state change to everyone except the sender.
+func (h *Hub) PublishTyping(conversationID, userID int64, isTyping bool) {
+	evt := Typing{ConversationID: conversationID, UserID: userID, IsTyping: isTyping}
+	h.sendToConversation(conversationID, envelope{Type: EventTyping, Data: mustMarshal(evt)}, userID)
+}
+
+// PublishRead delivers a read receipt to everyone except the reader.
+func (h *Hub) PublishRead(conversationID, userID, messageID int64) {
+	evt := Read{ConversationID: conversationID, UserID: userID, MessageID: messageID}
+	h.sendToConversation(conversationID, envelope{Type: EventRead, Data: mustMarshal(evt)}, userID)
+}
+
+func (h *Hub) register(c *client, conversationIDs []int64) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	if h.conns[c.userID] == nil {
+		h.conns[c.userID] = make(map[*client]struct{})
+	}
+	h.conns[c.userID][c] = struct{}{}
+	for _, cid := range conversationIDs {
+		if h.subs[cid] == nil {
+			h.subs[cid] = make(map[int64]struct{})
+		}
+		h.subs[cid][c.userID] = struct{}{}
+	}
+}
+
+func (h *Hub) unregister(c *client, conversationIDs []int64) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	if conns := h.conns[c.userID]; conns != nil {
+		delete(conns, c)
+		if len(conns) == 0 {
+			delete(h.conns, c.userID)
+		}
+	}
+	for _, cid := range conversationIDs {
+		if users := h.subs[cid]; users != nil {
+			delete(users, c.userID)
+			if len(users) == 0 {
+				delete(h.subs, cid)
+			}
+		}
+	}
+	close(c.done)
+}
+
+func (h *Hub) broadcastPresence(userID int64, conversationIDs []int64, status string) {
+	evt := Presence{UserID: userID, Status: status}
+	env := envelope{Type: EventPresence, Data: mustMarshal(evt)}
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	for _, cid := range conversationIDs {
+		for uid := range h.subs[cid] {
+			if uid == userID {
+				continue
+			}
+			for c := range h.conns[uid] {
+				c.enqueue(env)
+			}
+		}
+	}
+}
+
+func (h *Hub) sendToConversation(conversationID int64, env envelope, exceptUserID int64) {
+	payload, err := json.Marshal(env)
+	if err != nil {
+		slog.Error("ws: marshal event", "error", err)
+		return
+	}
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	for uid := range h.subs[conversationID] {
+		if uid == exceptUserID {
+			continue
+		}
+		for c := range h.conns[uid] {
+			c.enqueueRaw(payload)
+		}
+	}
+}
+
+func mustMarshal(v any) json.RawMessage {
+	data, err := json.Marshal(v)
+	if err != nil {
+		panic(err)
+	}
+	return data
+}

server/internal/realtime/hub_test.go

@@ -0,0 +1,262 @@
+package realtime
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/coder/websocket"
+)
+
+type fakeMembers struct {
+	mu        sync.Mutex
+	byUser    map[int64][]int64
+	member    map[[2]int64]bool
+	lastReads [][2]int64
+}
+
+func newFakeMembers() *fakeMembers {
+	return &fakeMembers{
+		byUser: make(map[int64][]int64),
+		member: make(map[[2]int64]bool),
+	}
+}
+
+func (f *fakeMembers) ConversationIDsForUser(_ context.Context, userID int64) ([]int64, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.byUser[userID], nil
+}
+
+func (f *fakeMembers) IsMember(_ context.Context, conversationID, userID int64) (bool, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.member[[2]int64{conversationID, userID}], nil
+}
+
+func (f *fakeMembers) UpdateLastRead(_ context.Context, conversationID, userID int64) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.lastReads = append(f.lastReads, [2]int64{conversationID, userID})
+	return nil
+}
+
+func (f *fakeMembers) readCount() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return len(f.lastReads)
+}
+
+func (f *fakeMembers) readAt(i int) [2]int64 {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.lastReads[i]
+}
+
+func waitFor(t *testing.T, cond func() bool, what string) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if cond() {
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatalf("timed out waiting for %s", what)
+}
+
+func startHub(t *testing.T, fm *fakeMembers) (*Hub, *httptest.Server) {
+	t.Helper()
+	hub := NewHub(fm)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		userID := int64(0)
+		switch r.URL.Query().Get("user") {
+		case "1":
+			userID = 1
+		case "2":
+			userID = 2
+		}
+		hub.HandleWS(w, r, userID)
+	}))
+	t.Cleanup(srv.Close)
+	return hub, srv
+}
+
+func dialWS(t *testing.T, srv *httptest.Server, user string) *websocket.Conn {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	url := "ws" + strings.TrimPrefix(srv.URL, "http") + "?user=" + user
+	conn, _, err := websocket.Dial(ctx, url, nil)
+	if err != nil {
+		t.Fatalf("websocket dial: %v", err)
+	}
+	t.Cleanup(func() { conn.Close(websocket.StatusNormalClosure, "") })
+	time.Sleep(50 * time.Millisecond)
+	return conn
+}
+
+func readEnvelope(t *testing.T, conn *websocket.Conn) envelope {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	_, data, err := conn.Read(ctx)
+	if err != nil {
+		t.Fatalf("ws read: %v", err)
+	}
+	var env envelope
+	if err := json.Unmarshal(data, &env); err != nil {
+		t.Fatalf("unmarshal envelope: %v", err)
+	}
+	return env
+}
+
+// readEventOfType reads events until one of the wanted type arrives,
+// skipping unrelated events such as presence.
+func readEventOfType(t *testing.T, conn *websocket.Conn, wantType string) envelope {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		env := readEnvelope(t, conn)
+		if env.Type == wantType {
+			return env
+		}
+	}
+	t.Fatalf("timed out waiting for event %q", wantType)
+	return envelope{}
+}
+
+func sendEnvelope(t *testing.T, conn *websocket.Conn, env envelope) {
+	t.Helper()
+	data, err := json.Marshal(env)
+	if err != nil {
+		t.Fatalf("marshal envelope: %v", err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	if err := conn.Write(ctx, websocket.MessageText, data); err != nil {
+		t.Fatalf("ws write: %v", err)
+	}
+}
+
+func TestMessageCreatedReachesSubscribers(t *testing.T) {
+	fm := newFakeMembers()
+	fm.byUser[1] = []int64{10}
+	fm.byUser[2] = []int64{10}
+	fm.member[[2]int64{10, 1}] = true
+	fm.member[[2]int64{10, 2}] = true
+
+	hub, srv := startHub(t, fm)
+	conn1 := dialWS(t, srv, "1")
+	conn2 := dialWS(t, srv, "2")
+
+	evt := MessageCreated{ConversationID: 10, MessageID: 55, SenderID: 1, Body: "hello"}
+	hub.PublishMessageCreated(10, evt)
+
+	for _, conn := range []*websocket.Conn{conn1, conn2} {
+		env := readEventOfType(t, conn, EventMessageCreated)
+		var got MessageCreated
+		if err := json.Unmarshal(env.Data, &got); err != nil {
+			t.Fatalf("unmarshal data: %v", err)
+		}
+		if got.MessageID != 55 || got.Body != "hello" {
+			t.Errorf("data = %+v, want MessageID=55 Body=hello", got)
+		}
+	}
+}
+
+func TestNonMemberTypingIsIgnored(t *testing.T) {
+	fm := newFakeMembers()
+	fm.byUser[1] = []int64{10}
+	fm.byUser[2] = []int64{10}
+	fm.member[[2]int64{10, 2}] = true
+
+	hub, srv := startHub(t, fm)
+	conn1 := dialWS(t, srv, "1")
+	conn2 := dialWS(t, srv, "2")
+
+	sendEnvelope(t, conn1, envelope{
+		Type: EventTyping,
+		Data: mustMarshal(clientTyping{ConversationID: 10, IsTyping: true}),
+	})
+
+	evt := MessageCreated{ConversationID: 10, MessageID: 1, SenderID: 2, Body: "x"}
+	hub.PublishMessageCreated(10, evt)
+	env := readEventOfType(t, conn2, EventMessageCreated)
+	if env.Type != EventMessageCreated {
+		t.Errorf("member got %q, want message.created (no typing leak)", env.Type)
+	}
+}
+
+func TestPresenceOnConnectAndClose(t *testing.T) {
+	fm := newFakeMembers()
+	fm.byUser[1] = []int64{10}
+	fm.byUser[2] = []int64{10}
+
+	_, srv := startHub(t, fm)
+	conn1 := dialWS(t, srv, "1")
+	conn2 := dialWS(t, srv, "2")
+
+	env := readEnvelope(t, conn1)
+	if env.Type != EventPresence {
+		t.Fatalf("first event type = %q, want presence", env.Type)
+	}
+	var p Presence
+	if err := json.Unmarshal(env.Data, &p); err != nil {
+		t.Fatalf("unmarshal presence: %v", err)
+	}
+	if p.UserID != 2 || p.Status != StatusOnline {
+		t.Errorf("presence = %+v, want user 2 online", p)
+	}
+
+	conn2.Close(websocket.StatusNormalClosure, "")
+	env = readEnvelope(t, conn1)
+	if env.Type != EventPresence {
+		t.Fatalf("closing event type = %q, want presence", env.Type)
+	}
+	if err := json.Unmarshal(env.Data, &p); err != nil {
+		t.Fatalf("unmarshal presence: %v", err)
+	}
+	if p.UserID != 2 || p.Status != StatusOffline {
+		t.Errorf("presence = %+v, want user 2 offline", p)
+	}
+}
+
+func TestReadEventUpdatesLastReadAndBroadcasts(t *testing.T) {
+	fm := newFakeMembers()
+	fm.byUser[1] = []int64{10}
+	fm.byUser[2] = []int64{10}
+	fm.member[[2]int64{10, 1}] = true
+	fm.member[[2]int64{10, 2}] = true
+
+	_, srv := startHub(t, fm)
+	conn1 := dialWS(t, srv, "1")
+	conn2 := dialWS(t, srv, "2")
+
+	sendEnvelope(t, conn1, envelope{
+		Type: EventRead,
+		Data: mustMarshal(clientRead{ConversationID: 10, MessageID: 99}),
+	})
+
+	waitFor(t, func() bool { return fm.readCount() == 1 }, "UpdateLastRead call")
+	if got := fm.readAt(0); got != [2]int64{10, 1} {
+		t.Errorf("lastReads[0] = %v, want [10 1]", got)
+	}
+
+	env := readEnvelope(t, conn2)
+	if env.Type != EventRead {
+		t.Fatalf("type = %q, want read", env.Type)
+	}
+	var r Read
+	if err := json.Unmarshal(env.Data, &r); err != nil {
+		t.Fatalf("unmarshal read: %v", err)
+	}
+	if r.UserID != 1 || r.MessageID != 99 || r.ConversationID != 10 {
+		t.Errorf("read = %+v, want user 1, message 99, conv 10", r)
+	}
+}