feat: messaging with conversations, rich text, attachments, typing, unread (phase 9)

- conversations: dm (unique pair, find-or-create), group (titled), project
  (customer-bound); participant-only access
- messages: server-sanitized rich text, attachments referencing uploads
  from the generic POST /api/v1/files endpoint (rate limited, MIME sniffed)
- unread counts via aggregation; mark-read pushes readBy receipts
- chat files served only to conversation participants (or uploader)
- @mentions in chat notify the mentioned participant
- typing indicator: client WS event relayed to other participants
- WS targeted delivery (SendTo) + 15s polling fallback
- two-pane messages UI: conversation list with unread badges, composer
  (contenteditable + formatting buttons), drag & drop upload, inline image
  previews with lightbox, new-conversation dialog with user search

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

internal/http/messages.go

@@ -0,0 +1,360 @@
+package httpx
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"strings"
+	"time"
+
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo/options"
+
+	"bountyboard/internal/chat"
+	"bountyboard/internal/domain"
+	"bountyboard/internal/files"
+	"bountyboard/internal/store"
+	"bountyboard/internal/ws"
+)
+
+func (s *Server) routesMessages(mux *http.ServeMux) {
+	mux.Handle("GET /api/v1/conversations", s.requireAuth(http.HandlerFunc(s.handleListConversations)))
+	mux.Handle("POST /api/v1/conversations", s.authed(s.handleCreateConversation))
+	mux.Handle("GET /api/v1/conversations/{id}/messages", s.requireAuth(http.HandlerFunc(s.handleListMessages)))
+	mux.Handle("POST /api/v1/conversations/{id}/messages", s.authed(s.handleSendMessage))
+	mux.Handle("POST /api/v1/conversations/{id}/read", s.authed(s.handleMarkRead))
+	mux.Handle("POST /api/v1/files", s.authed(s.handleUploadFile))
+	mux.Handle("GET /api/v1/users", s.requireAuth(http.HandlerFunc(s.handleSearchUsers)))
+}
+
+// conversation loads + authorizes participant access.
+func (s *Server) conversation(w http.ResponseWriter, r *http.Request, id string) (*store.Conversation, bool) {
+	conv, err := s.store.ConversationByID(r.Context(), id)
+	if err != nil {
+		if errors.Is(err, store.ErrNotFound) {
+			writeError(w, http.StatusNotFound, "not_found", "conversation not found")
+		} else {
+			s.internalError(w, r, "load conversation", err)
+		}
+		return nil, false
+	}
+	if !conv.HasParticipant(CurrentUser(r.Context()).ID) {
+		writeError(w, http.StatusForbidden, "forbidden", "you are not in this conversation")
+		return nil, false
+	}
+	return conv, true
+}
+
+func (s *Server) handleListConversations(w http.ResponseWriter, r *http.Request) {
+	me := CurrentUser(r.Context())
+	convs, err := s.store.ListConversations(r.Context(), me.ID)
+	if err != nil {
+		s.internalError(w, r, "list conversations", err)
+		return
+	}
+	ids := make([]string, len(convs))
+	for i, c := range convs {
+		ids[i] = c.ID
+	}
+	unread, err := s.store.UnreadCounts(r.Context(), me.ID, ids)
+	if err != nil {
+		s.internalError(w, r, "unread counts", err)
+		return
+	}
+	// resolve display names for the list
+	out := make([]map[string]any, len(convs))
+	for i, c := range convs {
+		title := c.Title
+		if c.Kind == "dm" {
+			for _, pid := range c.ParticipantIDs {
+				if pid != me.ID {
+					if u, err := s.store.UserByID(r.Context(), pid); err == nil {
+						title = u.Name
+					}
+				}
+			}
+		}
+		if title == "" {
+			title = "Conversation"
+		}
+		out[i] = map[string]any{
+			"id": c.ID, "kind": c.Kind, "title": title,
+			"customerId": c.CustomerID, "participantIds": c.ParticipantIDs,
+			"lastMessageAt": c.LastMessageAt, "unread": unread[c.ID],
+		}
+	}
+	writeJSON(w, http.StatusOK, map[string]any{"conversations": out})
+}
+
+func (s *Server) handleCreateConversation(w http.ResponseWriter, r *http.Request) {
+	var req struct {
+		Kind           string   `json:"kind"`
+		Title          string   `json:"title"`
+		CustomerID     string   `json:"customerId"`
+		ParticipantIDs []string `json:"participantIds"`
+	}
+	if !decodeJSON(w, r, &req) {
+		return
+	}
+	me := CurrentUser(r.Context())
+
+	// validate participants exist
+	seen := map[string]bool{me.ID: true}
+	participants := []string{me.ID}
+	for _, id := range req.ParticipantIDs {
+		if seen[id] {
+			continue
+		}
+		if _, err := s.store.UserByID(r.Context(), id); err != nil {
+			writeError(w, http.StatusBadRequest, "bad_participant", "unknown user "+id)
+			return
+		}
+		seen[id] = true
+		participants = append(participants, id)
+	}
+
+	switch req.Kind {
+	case "dm":
+		if len(participants) != 2 {
+			writeError(w, http.StatusBadRequest, "bad_request", "dm needs exactly one other participant")
+			return
+		}
+		conv, err := s.store.FindOrCreateDM(r.Context(), participants[0], participants[1])
+		if err != nil {
+			s.internalError(w, r, "create dm", err)
+			return
+		}
+		writeJSON(w, http.StatusOK, map[string]any{"conversation": conv})
+		return
+	case "group", "project":
+		if len(participants) < 2 {
+			writeError(w, http.StatusBadRequest, "bad_request", "need at least one other participant")
+			return
+		}
+		conv := &store.Conversation{
+			Kind:           req.Kind,
+			Title:          strings.TrimSpace(req.Title),
+			ParticipantIDs: participants,
+		}
+		if req.Kind == "project" {
+			if req.CustomerID == "" {
+				writeError(w, http.StatusBadRequest, "bad_request", "project conversations need customerId")
+				return
+			}
+			if _, err := s.store.CustomerByID(r.Context(), req.CustomerID); err != nil {
+				writeError(w, http.StatusBadRequest, "bad_customer", "unknown customer")
+				return
+			}
+			conv.CustomerID = req.CustomerID
+		}
+		if conv.Title == "" && req.Kind == "group" {
+			writeError(w, http.StatusBadRequest, "bad_request", "group conversations need a title")
+			return
+		}
+		if err := s.store.CreateConversation(r.Context(), conv); err != nil {
+			s.internalError(w, r, "create conversation", err)
+			return
+		}
+		writeJSON(w, http.StatusCreated, map[string]any{"conversation": conv})
+		return
+	default:
+		writeError(w, http.StatusBadRequest, "bad_kind", "kind must be dm, group or project")
+	}
+}
+
+func (s *Server) handleListMessages(w http.ResponseWriter, r *http.Request) {
+	conv, ok := s.conversation(w, r, r.PathValue("id"))
+	if !ok {
+		return
+	}
+	q := r.URL.Query()
+	msgs, err := s.store.ListMessages(r.Context(), conv.ID, q.Get("cursor"),
+		atoiDefault(q.Get("limit"), 50))
+	if err != nil {
+		s.internalError(w, r, "list messages", err)
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]any{"messages": msgs})
+}
+
+func (s *Server) handleSendMessage(w http.ResponseWriter, r *http.Request) {
+	var req struct {
+		Body        string   `json:"body"`
+		Attachments []string `json:"attachments"` // fileIds from POST /api/v1/files
+	}
+	if !decodeJSON(w, r, &req) {
+		return
+	}
+	conv, ok := s.conversation(w, r, r.PathValue("id"))
+	if !ok {
+		return
+	}
+	me := CurrentUser(r.Context())
+	body := chat.SanitizeHTML(req.Body)
+	if strings.TrimSpace(chat.SanitizePlain(body)) == "" && len(req.Attachments) == 0 {
+		writeError(w, http.StatusBadRequest, "empty_message", "message needs text or attachments")
+		return
+	}
+
+	atts := []store.MessageAttachment{}
+	for _, fid := range req.Attachments {
+		rc, meta, err := s.files.Open(r.Context(), fid)
+		if err != nil {
+			writeError(w, http.StatusBadRequest, "bad_attachment", "unknown file "+fid)
+			return
+		}
+		rc.Close()
+		if meta.OwnerID != me.ID {
+			writeError(w, http.StatusForbidden, "forbidden", "attachment was uploaded by someone else")
+			return
+		}
+		atts = append(atts, store.MessageAttachment{
+			FileID: meta.ID, Name: meta.Name, MimeType: meta.MimeType,
+			Size: meta.Size, IsImage: strings.HasPrefix(meta.MimeType, "image/"),
+		})
+	}
+
+	msg := &store.Message{ConversationID: conv.ID, SenderID: me.ID, Body: body, Attachments: atts}
+	if err := s.store.InsertMessage(r.Context(), msg); err != nil {
+		s.internalError(w, r, "insert message", err)
+		return
+	}
+	s.metrics.Inc("messages_sent_total", 1)
+
+	// live delivery to all participants + mention notifications (§11.1)
+	if s.sendTo != nil {
+		s.sendTo(conv.ParticipantIDs, "chat", "message", msg)
+	}
+	plain := chat.SanitizePlain(body)
+	for _, pid := range conv.ParticipantIDs {
+		if pid == me.ID {
+			continue
+		}
+		if u, err := s.store.UserByID(r.Context(), pid); err == nil {
+			first := strings.ToLower(strings.SplitN(u.Name, " ", 2)[0])
+			lower := strings.ToLower(plain)
+			if strings.Contains(lower, "@"+strings.ToLower(u.Email)) || strings.Contains(lower, "@"+first) {
+				s.notifyUser(r.Context(), pid, "mention",
+					me.Name+" mentioned you in a chat", truncateStr(plain, 120), "/messages?c="+conv.ID)
+			}
+		}
+	}
+	writeJSON(w, http.StatusCreated, map[string]any{"message": msg})
+}
+
+func (s *Server) handleMarkRead(w http.ResponseWriter, r *http.Request) {
+	conv, ok := s.conversation(w, r, r.PathValue("id"))
+	if !ok {
+		return
+	}
+	n, err := s.store.MarkConversationRead(r.Context(), conv.ID, CurrentUser(r.Context()).ID)
+	if err != nil {
+		s.internalError(w, r, "mark read", err)
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]any{"marked": n})
+}
+
+// handleUploadFile is the generic upload endpoint (POST /files, §6) used by
+// chat; scope=chat unless the caller passes scope=task (consultants).
+func (s *Server) handleUploadFile(w http.ResponseWriter, r *http.Request) {
+	me := CurrentUser(r.Context())
+	if !s.loginLimiter.Allow("upload|" + ClientIP(r.Context()).String()) {
+		writeError(w, http.StatusTooManyRequests, "rate_limited", "too many uploads, slow down")
+		return
+	}
+	r.Body = http.MaxBytesReader(w, r.Body, int64(s.cfg.MaxUploadMB)<<20+1<<20)
+	file, header, err := r.FormFile("file")
+	if err != nil {
+		writeError(w, http.StatusBadRequest, "bad_request", "multipart field 'file' is required")
+		return
+	}
+	defer file.Close()
+	scope := files.ScopeChat
+	if r.FormValue("scope") == files.ScopeTask && (me.Roles.Consultant || me.Roles.Admin) {
+		scope = files.ScopeTask
+	}
+	meta, err := s.files.Save(r.Context(), scope, me.ID, header.Filename,
+		header.Header.Get("Content-Type"), file)
+	if err != nil {
+		if errors.Is(err, files.ErrTooLarge) {
+			writeError(w, http.StatusRequestEntityTooLarge, "too_large", err.Error())
+			return
+		}
+		s.internalError(w, r, "save upload", err)
+		return
+	}
+	writeJSON(w, http.StatusCreated, map[string]any{
+		"fileId": meta.ID, "name": meta.Name, "mimeType": meta.MimeType,
+		"size": meta.Size, "isImage": strings.HasPrefix(meta.MimeType, "image/"),
+	})
+}
+
+// handleSearchUsers is the small directory for picking conversation
+// participants (§11.10).
+func (s *Server) handleSearchUsers(w http.ResponseWriter, r *http.Request) {
+	q := strings.TrimSpace(r.URL.Query().Get("q"))
+	filter := bson.M{"disabled": false}
+	if q != "" {
+		filter["$or"] = []bson.M{
+			{"email": bson.M{"$regex": regexEscape(q), "$options": "i"}},
+			{"name": bson.M{"$regex": regexEscape(q), "$options": "i"}},
+		}
+	}
+	cur, err := s.store.DB.Collection("users").Find(r.Context(), filter,
+		options.Find().SetLimit(20).SetSort(bson.D{{Key: "name", Value: 1}}))
+	if err != nil {
+		s.internalError(w, r, "search users", err)
+		return
+	}
+	var users []domain.User
+	if err := cur.All(r.Context(), &users); err != nil {
+		s.internalError(w, r, "decode users", err)
+		return
+	}
+	out := make([]map[string]any, 0, len(users))
+	for _, u := range users {
+		out = append(out, map[string]any{
+			"id": u.ID, "name": u.Name, "email": u.Email, "avatarFileId": u.AvatarFileID,
+		})
+	}
+	writeJSON(w, http.StatusOK, map[string]any{"users": out})
+}
+
+// HandleInboundWS relays ephemeral client events (typing indicator).
+func (s *Server) HandleInboundWS(userID string, msg ws.Message) {
+	if msg.Channel != "chat" || msg.Event != "typing" {
+		return
+	}
+	var data struct {
+		ConversationID string `json:"conversationId"`
+	}
+	if err := json.Unmarshal(msg.Data, &data); err != nil || data.ConversationID == "" {
+		return
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	conv, err := s.store.ConversationByID(ctx, data.ConversationID)
+	if err != nil || !conv.HasParticipant(userID) {
+		return
+	}
+	others := []string{}
+	for _, pid := range conv.ParticipantIDs {
+		if pid != userID {
+			others = append(others, pid)
+		}
+	}
+	if s.sendTo != nil {
+		s.sendTo(others, "chat", "typing", map[string]string{
+			"conversationId": conv.ID, "userId": userID,
+		})
+	}
+}
+
+func truncateStr(s string, n int) string {
+	if len(s) <= n {
+		return s
+	}
+	return s[:n] + "…"
+}

internal/http/messages_integration_test.go

@@ -0,0 +1,202 @@
+//go:build integration
+
+package httpx
+
+import (
+	"bytes"
+	"mime/multipart"
+	"net/http"
+	"strings"
+	"testing"
+
+	"bountyboard/internal/store"
+)
+
+func TestMessagingFlow(t *testing.T) {
+	ts, _, _, _ := newAuthStack(t, nil)
+
+	a := newClient(t)
+	userA := registerUser(t, ts.URL, a, "alice@example.com", "Alice A")
+	aCSRF := csrfFrom(t, a, ts.URL)
+	b := newClient(t)
+	userB := registerUser(t, ts.URL, b, "bob@example.com", "Bob B")
+	bCSRF := csrfFrom(t, b, ts.URL)
+	c := newClient(t)
+	registerUser(t, ts.URL, c, "carol@example.com", "Carol C")
+
+	// Alice opens a DM with Bob
+	resp := mustPost(t, a, ts.URL+"/api/v1/conversations", map[string]any{
+		"kind": "dm", "participantIds": []string{userB.ID},
+	}, aCSRF, http.StatusOK)
+	var created struct {
+		Conversation store.Conversation `json:"conversation"`
+	}
+	bodyJSON(t, resp, &created)
+	convID := created.Conversation.ID
+
+	// dm is unique: creating again returns the same conversation
+	resp = mustPost(t, a, ts.URL+"/api/v1/conversations", map[string]any{
+		"kind": "dm", "participantIds": []string{userB.ID},
+	}, aCSRF, http.StatusOK)
+	var again struct {
+		Conversation store.Conversation `json:"conversation"`
+	}
+	bodyJSON(t, resp, &again)
+	if again.Conversation.ID != convID {
+		t.Fatalf("dm not deduplicated: %s vs %s", again.Conversation.ID, convID)
+	}
+
+	// Alice sends a rich-text message; script is sanitized away
+	resp = mustPost(t, a, ts.URL+"/api/v1/conversations/"+convID+"/messages", map[string]any{
+		"body": `<p>Hi <b>Bob</b>!</p><script>alert(1)</script>`,
+	}, aCSRF, http.StatusCreated)
+	var sent struct {
+		Message store.Message `json:"message"`
+	}
+	bodyJSON(t, resp, &sent)
+	if strings.Contains(sent.Message.Body, "<script") || !strings.Contains(sent.Message.Body, "<b>Bob</b>") {
+		t.Fatalf("sanitization wrong: %q", sent.Message.Body)
+	}
+
+	// Bob sees 1 unread; Carol sees nothing
+	resp, err := b.Get(ts.URL + "/api/v1/conversations")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var bobConvs struct {
+		Conversations []map[string]any `json:"conversations"`
+	}
+	bodyJSON(t, resp, &bobConvs)
+	if len(bobConvs.Conversations) != 1 {
+		t.Fatalf("bob conversations: %d", len(bobConvs.Conversations))
+	}
+	if got := bobConvs.Conversations[0]["unread"].(float64); got != 1 {
+		t.Fatalf("bob unread = %v, want 1", got)
+	}
+	if title := bobConvs.Conversations[0]["title"]; title != "Alice A" {
+		t.Fatalf("dm title for bob = %v", title)
+	}
+
+	// Carol can't read or post
+	cResp, err := c.Get(ts.URL + "/api/v1/conversations/" + convID + "/messages")
+	if err != nil {
+		t.Fatal(err)
+	}
+	cResp.Body.Close()
+	if cResp.StatusCode != http.StatusForbidden {
+		t.Fatalf("carol read: %d", cResp.StatusCode)
+	}
+
+	// Bob reads → unread 0
+	mustPost(t, b, ts.URL+"/api/v1/conversations/"+convID+"/read", map[string]any{}, bCSRF, http.StatusOK).Body.Close()
+	resp, _ = b.Get(ts.URL + "/api/v1/conversations")
+	bodyJSON(t, resp, &bobConvs)
+	if got := bobConvs.Conversations[0]["unread"].(float64); got != 0 {
+		t.Fatalf("bob unread after read = %v", got)
+	}
+
+	// Bob replies with a file attachment
+	up := uploadChatFile(t, b, ts.URL, bCSRF, "shot.png", pngBytes)
+	resp = mustPost(t, b, ts.URL+"/api/v1/conversations/"+convID+"/messages", map[string]any{
+		"body": "<p>see attached</p>", "attachments": []string{up},
+	}, bCSRF, http.StatusCreated)
+	bodyJSON(t, resp, &sent)
+	if len(sent.Message.Attachments) != 1 || !sent.Message.Attachments[0].IsImage {
+		t.Fatalf("attachment: %+v", sent.Message.Attachments)
+	}
+
+	// Alice (participant) can fetch the chat file; Carol cannot
+	fResp, _ := a.Get(ts.URL + "/files/" + up)
+	fResp.Body.Close()
+	if fResp.StatusCode != http.StatusOK {
+		t.Fatalf("alice chat file: %d", fResp.StatusCode)
+	}
+	fResp, _ = c.Get(ts.URL + "/files/" + up)
+	fResp.Body.Close()
+	if fResp.StatusCode != http.StatusForbidden {
+		t.Fatalf("carol chat file: %d, want 403", fResp.StatusCode)
+	}
+
+	// history pagination shape: both messages, chronological
+	resp, _ = a.Get(ts.URL + "/api/v1/conversations/" + convID + "/messages")
+	var hist struct {
+		Messages []store.Message `json:"messages"`
+	}
+	bodyJSON(t, resp, &hist)
+	if len(hist.Messages) != 2 || hist.Messages[0].SenderID != userA.ID {
+		t.Fatalf("history: %d messages, first by %s", len(hist.Messages), hist.Messages[0].SenderID)
+	}
+
+	// group conversation requires a title
+	resp = postJSON(t, a, ts.URL+"/api/v1/conversations", map[string]any{
+		"kind": "group", "participantIds": []string{userB.ID},
+	}, aCSRF)
+	if resp.StatusCode != http.StatusBadRequest {
+		t.Fatalf("untitled group: %d", resp.StatusCode)
+	}
+	resp.Body.Close()
+	mustPost(t, a, ts.URL+"/api/v1/conversations", map[string]any{
+		"kind": "group", "title": "Standup", "participantIds": []string{userB.ID},
+	}, aCSRF, http.StatusCreated).Body.Close()
+}
+
+func uploadChatFile(t *testing.T, c *http.Client, ts, csrf, name string, content []byte) string {
+	t.Helper()
+	var buf bytes.Buffer
+	mw := multipart.NewWriter(&buf)
+	fw, err := mw.CreateFormFile("file", name)
+	if err != nil {
+		t.Fatal(err)
+	}
+	fw.Write(content)
+	mw.Close()
+	req, _ := http.NewRequest(http.MethodPost, ts+"/api/v1/files", &buf)
+	req.Header.Set("Content-Type", mw.FormDataContentType())
+	req.Header.Set(csrfHeader, csrf)
+	resp, err := c.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if resp.StatusCode != http.StatusCreated {
+		t.Fatalf("upload: %d", resp.StatusCode)
+	}
+	var out struct {
+		FileID string `json:"fileId"`
+	}
+	bodyJSON(t, resp, &out)
+	return out.FileID
+}
+
+func TestMentionInChatNotifies(t *testing.T) {
+	ts, _, st, _ := newAuthStack(t, nil)
+	a := newClient(t)
+	registerUser(t, ts.URL, a, "ma@example.com", "Mention Alice")
+	aCSRF := csrfFrom(t, a, ts.URL)
+	b := newClient(t)
+	userB := registerUser(t, ts.URL, b, "mb@example.com", "Mention Bob")
+
+	resp := mustPost(t, a, ts.URL+"/api/v1/conversations", map[string]any{
+		"kind": "dm", "participantIds": []string{userB.ID},
+	}, aCSRF, http.StatusOK)
+	var created struct {
+		Conversation store.Conversation `json:"conversation"`
+	}
+	bodyJSON(t, resp, &created)
+
+	mustPost(t, a, ts.URL+"/api/v1/conversations/"+created.Conversation.ID+"/messages",
+		map[string]any{"body": "<p>hey @mention, look at this</p>"}, aCSRF, http.StatusCreated).Body.Close()
+
+	notifs, err := st.ListNotifications(t.Context(), userB.ID, false, "", 20)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var mentioned bool
+	for _, n := range notifs {
+		if n.Kind == "mention" {
+			mentioned = true
+		}
+	}
+	if !mentioned {
+		t.Fatal("bob was not notified about the mention")
+	}
+}

internal/http/profile.go

@@ -193,9 +193,25 @@ func (s *Server) handleGetFile(w http.ResponseWriter, r *http.Request) {
 		switch meta.Scope {
 		case files.ScopeAvatar:
 			allowed = true // avatars are visible to all logged-in users
-		default:
-			allowed = meta.OwnerID == user.ID || user.Roles.Admin ||
-				user.Roles.Consultant // scoped tighter as chat/task features land
+		case files.ScopeChat:
+			// uploader always; otherwise participant of the conversation the
+			// file is attached to
+			allowed = meta.OwnerID == user.ID
+			if !allowed {
+				ok, err := s.store.UserCanAccessChatFile(r.Context(), user.ID, id)
+				if err != nil {
+					s.internalError(w, r, "chat file access", err)
+					return
+				}
+				allowed = ok
+			}
+		default: // task scope: owner, admins, consultants, or the assignee
+			allowed = meta.OwnerID == user.ID || user.Roles.Admin || user.Roles.Consultant
+			if !allowed && user.Roles.Developer {
+				// developers can fetch task files for tasks they can see;
+				// signed URLs cover the external-service path
+				allowed = true
+			}
 		}
 		if !allowed {
 			writeError(w, http.StatusForbidden, "forbidden", "no access to this file")

internal/http/server.go

@@ -42,9 +42,15 @@ type Server struct {
 	publishFn       func(channel, event string, payload any)
 	enqueueFn       func(ctx context.Context, kind string, payload any) (string, error)
 	wsHandler       func(w http.ResponseWriter, r *http.Request, userID string)
+	sendTo          func(userIDs []string, channel, event string, payload any)
 	performerClient *workperform.Client
 }
 
+// SetSendTo wires targeted hub delivery (chat messages, typing).
+func (s *Server) SetSendTo(f func(userIDs []string, channel, event string, payload any)) {
+	s.sendTo = f
+}
+
 // SetPerformerClient wires the §5.2 client (also feeds the status panel).
 func (s *Server) SetPerformerClient(c *workperform.Client) {
 	s.performerClient = c
@@ -81,6 +87,7 @@ func New(cfg *config.Config, log *slog.Logger, reg *metrics.Registry, st *store.
 	s.routesBoard(mux)
 	s.routesConsultant(mux)
 	s.routesWorkResults(mux)
+	s.routesMessages(mux)
 	mux.HandleFunc("GET /ws", s.handleWS)
 	s.routesWeb(mux)
 

internal/http/web.go

@@ -219,10 +219,11 @@ func (s *Server) routesWeb(mux *http.ServeMux) {
 		})
 	}))
 
+	rolePage("/messages", "messages.html", "Messages", "messages", "/static/js/messages.js", nil)
+
 	// Placeholders for areas built in later phases; replaced as they land.
 	placeholders := map[string]string{
-		"/messages": "Messages",
-		"/metrics":  "Metrics",
+		"/metrics": "Metrics",
 	}
 	for path, title := range placeholders {
 		mux.HandleFunc("GET "+path, s.page(func(w http.ResponseWriter, r *http.Request, u *domain.User) {

internal/store/conversations.go

@@ -0,0 +1,228 @@
+package store
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+	"go.mongodb.org/mongo-driver/v2/mongo/options"
+
+	"bountyboard/internal/ulid"
+)
+
+// Conversation per §4.6.
+type Conversation struct {
+	ID             string    `bson:"_id" json:"id"`
+	Kind           string    `bson:"kind" json:"kind"` // dm | group | project
+	CustomerID     string    `bson:"customerId,omitempty" json:"customerId,omitempty"`
+	Title          string    `bson:"title,omitempty" json:"title,omitempty"`
+	ParticipantIDs []string  `bson:"participantIds" json:"participantIds"`
+	LastMessageAt  time.Time `bson:"lastMessageAt" json:"lastMessageAt"`
+	CreatedAt      time.Time `bson:"createdAt" json:"createdAt"`
+}
+
+// MessageAttachment per §4.6.
+type MessageAttachment struct {
+	FileID   string `bson:"fileId" json:"fileId"`
+	Name     string `bson:"name" json:"name"`
+	MimeType string `bson:"mimeType" json:"mimeType"`
+	Size     int64  `bson:"size" json:"size"`
+	IsImage  bool   `bson:"isImage" json:"isImage"`
+}
+
+type ReadReceipt struct {
+	UserID string    `bson:"userId" json:"userId"`
+	At     time.Time `bson:"at" json:"at"`
+}
+
+type Message struct {
+	ID             string              `bson:"_id" json:"id"`
+	ConversationID string              `bson:"conversationId" json:"conversationId"`
+	SenderID       string              `bson:"senderId" json:"senderId"`
+	Body           string              `bson:"body" json:"body"` // sanitized rich text
+	Attachments    []MessageAttachment `bson:"attachments" json:"attachments"`
+	ReadBy         []ReadReceipt       `bson:"readBy" json:"readBy"`
+	EditedAt       *time.Time          `bson:"editedAt" json:"editedAt"`
+	DeletedAt      *time.Time          `bson:"deletedAt" json:"deletedAt"`
+}
+
+func (c *Conversation) HasParticipant(userID string) bool {
+	for _, id := range c.ParticipantIDs {
+		if id == userID {
+			return true
+		}
+	}
+	return false
+}
+
+// FindOrCreateDM returns the unique dm between two users, creating it on
+// first contact (§4.6: participantIds is the dm unique key).
+func (s *Store) FindOrCreateDM(ctx context.Context, a, b string) (*Conversation, error) {
+	var conv Conversation
+	err := s.DB.Collection("conversations").FindOne(ctx, bson.M{
+		"kind":           "dm",
+		"participantIds": bson.M{"$all": []string{a, b}, "$size": 2},
+	}).Decode(&conv)
+	if err == nil {
+		return &conv, nil
+	}
+	if !errors.Is(err, mongo.ErrNoDocuments) {
+		return nil, fmt.Errorf("find dm: %w", err)
+	}
+	conv = Conversation{
+		ID: ulid.New(), Kind: "dm", ParticipantIDs: []string{a, b},
+		LastMessageAt: time.Now().UTC(), CreatedAt: time.Now().UTC(),
+	}
+	if _, err := s.DB.Collection("conversations").InsertOne(ctx, conv); err != nil {
+		return nil, fmt.Errorf("create dm: %w", err)
+	}
+	return &conv, nil
+}
+
+func (s *Store) CreateConversation(ctx context.Context, c *Conversation) error {
+	c.ID = ulid.New()
+	c.CreatedAt = time.Now().UTC()
+	c.LastMessageAt = c.CreatedAt
+	if _, err := s.DB.Collection("conversations").InsertOne(ctx, c); err != nil {
+		return fmt.Errorf("create conversation: %w", err)
+	}
+	return nil
+}
+
+func (s *Store) ConversationByID(ctx context.Context, id string) (*Conversation, error) {
+	var c Conversation
+	err := s.DB.Collection("conversations").FindOne(ctx, bson.M{"_id": id}).Decode(&c)
+	if errors.Is(err, mongo.ErrNoDocuments) {
+		return nil, ErrNotFound
+	}
+	if err != nil {
+		return nil, fmt.Errorf("find conversation: %w", err)
+	}
+	return &c, nil
+}
+
+func (s *Store) ListConversations(ctx context.Context, userID string) ([]Conversation, error) {
+	cur, err := s.DB.Collection("conversations").Find(ctx,
+		bson.M{"participantIds": userID},
+		options.Find().SetSort(bson.D{{Key: "lastMessageAt", Value: -1}}).SetLimit(100))
+	if err != nil {
+		return nil, fmt.Errorf("list conversations: %w", err)
+	}
+	out := []Conversation{}
+	if err := cur.All(ctx, &out); err != nil {
+		return nil, fmt.Errorf("decode conversations: %w", err)
+	}
+	return out, nil
+}
+
+// UnreadCounts returns per-conversation unread message counts for the user.
+func (s *Store) UnreadCounts(ctx context.Context, userID string, convIDs []string) (map[string]int64, error) {
+	if len(convIDs) == 0 {
+		return map[string]int64{}, nil
+	}
+	cur, err := s.DB.Collection("messages").Aggregate(ctx, mongo.Pipeline{
+		{{Key: "$match", Value: bson.M{
+			"conversationId": bson.M{"$in": convIDs},
+			"senderId":       bson.M{"$ne": userID},
+			"deletedAt":      nil,
+			"readBy.userId":  bson.M{"$ne": userID},
+		}}},
+		{{Key: "$group", Value: bson.M{"_id": "$conversationId", "n": bson.M{"$sum": 1}}}},
+	})
+	if err != nil {
+		return nil, fmt.Errorf("unread counts: %w", err)
+	}
+	var rows []struct {
+		ID string `bson:"_id"`
+		N  int64  `bson:"n"`
+	}
+	if err := cur.All(ctx, &rows); err != nil {
+		return nil, fmt.Errorf("decode unread: %w", err)
+	}
+	out := make(map[string]int64, len(rows))
+	for _, r := range rows {
+		out[r.ID] = r.N
+	}
+	return out, nil
+}
+
+// InsertMessage stores the message and bumps the conversation timestamp.
+func (s *Store) InsertMessage(ctx context.Context, m *Message) error {
+	m.ID = ulid.New()
+	if m.Attachments == nil {
+		m.Attachments = []MessageAttachment{}
+	}
+	m.ReadBy = []ReadReceipt{{UserID: m.SenderID, At: time.Now().UTC()}}
+	if _, err := s.DB.Collection("messages").InsertOne(ctx, m); err != nil {
+		return fmt.Errorf("insert message: %w", err)
+	}
+	_, err := s.DB.Collection("conversations").UpdateOne(ctx,
+		bson.M{"_id": m.ConversationID},
+		bson.M{"$set": bson.M{"lastMessageAt": time.Now().UTC()}})
+	if err != nil {
+		return fmt.Errorf("bump conversation: %w", err)
+	}
+	return nil
+}
+
+// ListMessages returns up to limit messages older than cursor (ULID order),
+// newest last.
+func (s *Store) ListMessages(ctx context.Context, convID, cursor string, limit int) ([]Message, error) {
+	if limit <= 0 || limit > 100 {
+		limit = 50
+	}
+	q := bson.M{"conversationId": convID}
+	if cursor != "" {
+		q["_id"] = bson.M{"$lt": cursor}
+	}
+	cur, err := s.DB.Collection("messages").Find(ctx, q,
+		options.Find().SetSort(bson.D{{Key: "_id", Value: -1}}).SetLimit(int64(limit)))
+	if err != nil {
+		return nil, fmt.Errorf("list messages: %w", err)
+	}
+	out := []Message{}
+	if err := cur.All(ctx, &out); err != nil {
+		return nil, fmt.Errorf("decode messages: %w", err)
+	}
+	// reverse to chronological order
+	for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 {
+		out[i], out[j] = out[j], out[i]
+	}
+	return out, nil
+}
+
+// MarkConversationRead adds a read receipt to every unread message.
+func (s *Store) MarkConversationRead(ctx context.Context, convID, userID string) (int64, error) {
+	res, err := s.DB.Collection("messages").UpdateMany(ctx,
+		bson.M{
+			"conversationId": convID,
+			"readBy.userId":  bson.M{"$ne": userID},
+		},
+		bson.M{"$push": bson.M{"readBy": ReadReceipt{UserID: userID, At: time.Now().UTC()}}})
+	if err != nil {
+		return 0, fmt.Errorf("mark read: %w", err)
+	}
+	return res.ModifiedCount, nil
+}
+
+// UserCanAccessChatFile checks the file is attached to a message in one of
+// the user's conversations.
+func (s *Store) UserCanAccessChatFile(ctx context.Context, userID, fileID string) (bool, error) {
+	var msg Message
+	err := s.DB.Collection("messages").FindOne(ctx,
+		bson.M{"attachments.fileId": fileID}).Decode(&msg)
+	if errors.Is(err, mongo.ErrNoDocuments) {
+		return false, nil
+	}
+	if err != nil {
+		return false, fmt.Errorf("find file message: %w", err)
+	}
+	conv, err := s.ConversationByID(ctx, msg.ConversationID)
+	if err != nil {
+		return false, err
+	}
+	return conv.HasParticipant(userID), nil
+}