diff --git a/PROGRESS.md b/PROGRESS.md index 4a6c111..73aa979 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -9,3 +9,4 @@ - Phase 6 (sync workers): per-customer pollers with reconcile loop (start/stop/interval changes), §5.3 idempotent upsert keyed (system,key,customerId) w/ content-hash change detection, upstream edits refresh only while imported (timeline+notification after), attachment caching to GridFS, orphan flagging (never deletes), admin sync-now trigger + worker statuses, default budget prefill — demo-type integration tests green. - Phase 7 (atomization): circuit breaker (3 fails→open, 60s half-open probe) + retrying bearer JSON client shared by both external services, atomizer client w/ §5.1 coefficient normalization (±0.001 ok, ≤0.05 renormalized, else 502-class), persisted jobs queue (panic-safe, backoff ×3, stale requeue, restart-safe), subdivide/extend endpoints (202 async, re-run replaces unpublished children after confirm), task editing w/ bounty recompute + optional budget cascade, publish single/bulk, WS hub (origin check, 30s heartbeats) + live board, consultant atomization board UI (tree, sliders, sum indicator, modals, shimmer, health gating) — unit + integration green. Fixed tasks unique index: sparse→partial (sparse compound matched every task via customerId). - Phase 8 (bounty board + lifecycle): developer board (pool-scoped, filters/search/sort, stale age badges, hide-competing-claims setting, saved filters), claim/withdraw/decline/approve, assign-to-AI via §5.2 client + HMAC-verified idempotent callback w/ artifact ingestion into GridFS, start/submit/abandon/unassign, comments (sanitized, @mentions→notifications), time log, review queue + per-AC checklist on timeline, immutable bountyAwards on approve (humans only), notifications center + bell + WS toasts, my-tasks kanban, task detail page, pool management UI, server-side HTML sanitizer w/ XSS vector tests — full lifecycle integration tests green. +- Phase 9 (messaging): conversations (dm dedupe/group/project), sanitized rich-text messages, file/image attachments (generic POST /api/v1/files w/ rate limit), per-conversation unread counts + mark-read, chat-file access limited to participants, mentions→notifications, typing indicator over WS inbound relay, two-pane messages UI (contenteditable composer, formatting toolbar, drag&drop upload, inline previews + lightbox, live delivery, polling fallback), user directory search — integration tests green. diff --git a/cmd/app/main.go b/cmd/app/main.go index 411010e..40df7df 100644 --- a/cmd/app/main.go +++ b/cmd/app/main.go @@ -86,6 +86,8 @@ func run() error { }) srv.SetWSHandler(hub.Handle) srv.SetPublishFn(hub.Broadcast) + srv.SetSendTo(hub.SendTo) + hub.SetInbound(srv.HandleInboundWS) // Atomizer client honoring the admin base-URL override per call. atomClient := atomize.New(func() string { diff --git a/internal/http/messages.go b/internal/http/messages.go new file mode 100644 index 0000000..cdca030 --- /dev/null +++ b/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] + "…" +} diff --git a/internal/http/messages_integration_test.go b/internal/http/messages_integration_test.go new file mode 100644 index 0000000..4dabc74 --- /dev/null +++ b/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": `
Hi Bob!
`, + }, aCSRF, http.StatusCreated) + var sent struct { + Message store.Message `json:"message"` + } + bodyJSON(t, resp, &sent) + if strings.Contains(sent.Message.Body, "