feat: bounty board, task lifecycle, AI work performer flow, notifications (phase 8)

- whitelist HTML sanitizer (stdlib tokenizer) with XSS vector tests
- developer board: pool-scoped visibility, customer/search/minBounty/sort
  filters, stale-task age badges, competing-claims visibility setting,
  saved filters in profile extras
- claims: request/withdraw (developer), approve/decline (consultant) with
  notifications to winners and losers; unassign/abandon back to board
- work tracking: start, sanitized comments with @mention notifications,
  time logging, submit for review
- review queue + review with per-AC checklist stored on the timeline;
  approve writes the immutable bountyAwards row (human assignees only)
- assign-to-AI: §5.2 job submission, HMAC-verified callback endpoint,
  idempotent by jobId, artifacts downloaded into GridFS, failure path
  keeps the task assigned with timeline + notification
- notifications API + bell with unread badge, dropdown, page, WS toasts
- pages: bounty board, my-tasks kanban, task detail (role-driven actions,
  review dialog, AI dialog), review queue, developer pool

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
etalon
2026-06-12 20:11:05 +02:00
parent f87b954f27
commit 70a813edfa
28 changed files with 2764 additions and 14 deletions
+486
View File
@@ -0,0 +1,486 @@
package httpx
import (
"context"
"errors"
"net/http"
"regexp"
"strconv"
"strings"
"time"
"go.mongodb.org/mongo-driver/v2/bson"
"bountyboard/internal/chat"
"bountyboard/internal/domain"
"bountyboard/internal/store"
"bountyboard/internal/ulid"
)
func (s *Server) routesBoard(mux *http.ServeMux) {
dev := func(h http.HandlerFunc) http.Handler { return s.authedRole("developer", h) }
mux.Handle("GET /api/v1/board", s.requireAuth(s.requireRole("developer", http.HandlerFunc(s.handleBoard))))
mux.Handle("GET /api/v1/my-tasks", s.requireAuth(s.requireRole("developer", http.HandlerFunc(s.handleMyTasks))))
mux.Handle("POST /api/v1/tasks/{id}/claim", dev(s.handleClaim))
mux.Handle("POST /api/v1/tasks/{id}/claim/withdraw", dev(s.handleWithdrawClaim))
mux.Handle("POST /api/v1/tasks/{id}/start", dev(s.handleStartTask))
mux.Handle("POST /api/v1/tasks/{id}/submit-review", dev(s.handleSubmitReview))
mux.Handle("POST /api/v1/tasks/{id}/abandon", dev(s.handleAbandonTask))
mux.Handle("POST /api/v1/tasks/{id}/comments", s.authed(s.handleAddComment))
mux.Handle("POST /api/v1/tasks/{id}/time", dev(s.handleLogTime))
mux.Handle("GET /api/v1/notifications", s.requireAuth(http.HandlerFunc(s.handleListNotifications)))
mux.Handle("POST /api/v1/notifications/read", s.authed(s.handleReadNotifications))
mux.Handle("GET /api/v1/users/{id}/card", s.requireAuth(http.HandlerFunc(s.handleUserCard)))
}
// boardConsultants resolves which consultants' tasks the developer sees:
// the consultants who selected them into a pool (§3).
func (s *Server) boardConsultants(r *http.Request) ([]string, error) {
return s.store.PoolConsultantIDs(r.Context(), CurrentUser(r.Context()).ID)
}
func (s *Server) handleBoard(w http.ResponseWriter, r *http.Request) {
consultants, err := s.boardConsultants(r)
if err != nil {
s.internalError(w, r, "pool lookup", err)
return
}
q := r.URL.Query()
tasks := []domain.Task{}
if len(consultants) > 0 {
minBounty, _ := strconv.ParseFloat(q.Get("minBounty"), 64)
filter := store.TaskFilter{
ConsultantIDs: consultants,
Statuses: []domain.TaskStatus{domain.StatusPublished, domain.StatusClaimRequested},
Search: strings.TrimSpace(q.Get("q")),
MinBounty: minBounty,
CustomerID: q.Get("customerId"),
Sort: q.Get("sort"), // newest (default) | bounty
Limit: 200,
}
if tasks, err = s.store.ListTasks(r.Context(), filter); err != nil {
s.internalError(w, r, "list board", err)
return
}
}
// §10: optionally hide competing claim requests from other developers.
settings, err := s.store.GetSettings(r.Context())
hideCompeting := err == nil && settings.HideCompetingClaims
me := CurrentUser(r.Context()).ID
for i := range tasks {
mine := []domain.ClaimRequest{}
for _, cr := range tasks[i].ClaimRequests {
if cr.DeveloperID == me || !hideCompeting {
mine = append(mine, cr)
}
}
tasks[i].ClaimRequests = mine
}
// customers for the filter dropdown
customerIDs := map[string]bool{}
for _, t := range tasks {
customerIDs[t.CustomerID] = true
}
customers := []map[string]any{}
for id := range customerIDs {
if c, err := s.store.CustomerByID(r.Context(), id); err == nil {
customers = append(customers, map[string]any{"id": c.ID, "name": c.Name})
}
}
writeJSON(w, http.StatusOK, map[string]any{"tasks": tasks, "customers": customers})
}
func (s *Server) handleMyTasks(w http.ResponseWriter, r *http.Request) {
me := CurrentUser(r.Context()).ID
tasks, err := s.store.ListTasks(r.Context(), store.TaskFilter{
AssigneeID: me,
Statuses: []domain.TaskStatus{
domain.StatusAssigned, domain.StatusInProgress, domain.StatusInReview,
domain.StatusChangesRequested, domain.StatusApproved,
},
Sort: "updated",
Limit: 200,
})
if err != nil {
s.internalError(w, r, "list my tasks", err)
return
}
writeJSON(w, http.StatusOK, map[string]any{"tasks": tasks})
}
// developerTask loads a task ensuring the developer can see it via a pool
// relationship with its consultant.
func (s *Server) developerTask(w http.ResponseWriter, r *http.Request, id string) (*domain.Task, bool) {
t, err := s.store.TaskByID(r.Context(), id)
if err != nil {
if errors.Is(err, store.ErrNotFound) {
writeError(w, http.StatusNotFound, "not_found", "task not found")
} else {
s.internalError(w, r, "load task", err)
}
return nil, false
}
consultants, err := s.boardConsultants(r)
if err != nil {
s.internalError(w, r, "pool lookup", err)
return nil, false
}
for _, cid := range consultants {
if cid == t.ConsultantID {
return t, true
}
}
// assignees keep access even if removed from the pool mid-flight
if t.IsAssignedTo(CurrentUser(r.Context()).ID) {
return t, true
}
writeError(w, http.StatusForbidden, "forbidden", "this task is not on your board")
return nil, false
}
func (s *Server) handleClaim(w http.ResponseWriter, r *http.Request) {
var req struct {
Note string `json:"note"`
}
if !decodeJSON(w, r, &req) {
return
}
t, ok := s.developerTask(w, r, r.PathValue("id"))
if !ok {
return
}
me := CurrentUser(r.Context())
if t.HasClaimFrom(me.ID) {
writeError(w, http.StatusConflict, "already_requested", "you already requested this task")
return
}
claim := domain.ClaimRequest{DeveloperID: me.ID, Note: req.Note, At: time.Now().UTC()}
switch t.Status {
case domain.StatusPublished:
err := s.store.TransitionTask(r.Context(), t, domain.StatusClaimRequested,
me.ID, "claim_requested", map[string]any{"note": req.Note},
map[string]any{"claimRequests": []domain.ClaimRequest{claim}})
if err != nil {
s.taskTransitionError(w, err)
return
}
case domain.StatusClaimRequested:
// additional developers may pile on (visible per settings)
err := store.UpdateVersioned(r.Context(), s.store.DB.Collection("tasks"), t.ID, t.Version, bson.M{
"$push": bson.M{
"claimRequests": claim,
"timeline": domain.TimelineEntry{
At: time.Now().UTC(), ActorID: me.ID, Event: "claim_requested",
Data: map[string]any{"note": req.Note},
},
},
})
if err != nil {
s.taskTransitionError(w, err)
return
}
default:
writeError(w, http.StatusConflict, "bad_status", "task is not open for claims")
return
}
s.notifyUser(r.Context(), t.ConsultantID, "claim_requested",
"Assignment requested: "+t.Title, me.Name+" wants this task.", "/tasks/"+t.ID)
s.Publish("board", "task.claimed", map[string]any{"taskId": t.ID, "customerId": t.CustomerID})
w.WriteHeader(http.StatusNoContent)
}
func (s *Server) handleWithdrawClaim(w http.ResponseWriter, r *http.Request) {
t, ok := s.developerTask(w, r, r.PathValue("id"))
if !ok {
return
}
me := CurrentUser(r.Context())
if !t.HasClaimFrom(me.ID) {
writeError(w, http.StatusConflict, "no_claim", "you have no claim on this task")
return
}
s.removeClaim(w, r, t, me.ID, me.ID, "claim_withdrawn")
}
// removeClaim drops one developer's claim; the task returns to published
// when no claims remain (§4.4).
func (s *Server) removeClaim(w http.ResponseWriter, r *http.Request, t *domain.Task,
developerID, actorID, event string) {
remaining := []domain.ClaimRequest{}
for _, cr := range t.ClaimRequests {
if cr.DeveloperID != developerID {
remaining = append(remaining, cr)
}
}
update := bson.M{
"$set": bson.M{"claimRequests": remaining},
"$push": bson.M{"timeline": domain.TimelineEntry{
At: time.Now().UTC(), ActorID: actorID, Event: event,
Data: map[string]any{"developerId": developerID},
}},
}
if len(remaining) == 0 && t.Status == domain.StatusClaimRequested {
if err := domain.Transition(t.Status, domain.StatusPublished); err != nil {
s.taskTransitionError(w, err)
return
}
update["$set"].(bson.M)["status"] = domain.StatusPublished
}
if err := store.UpdateVersioned(r.Context(), s.store.DB.Collection("tasks"), t.ID, t.Version, update); err != nil {
s.taskTransitionError(w, err)
return
}
s.Publish("board", "task.claim_removed", map[string]any{"taskId": t.ID, "customerId": t.CustomerID})
w.WriteHeader(http.StatusNoContent)
}
func (s *Server) handleStartTask(w http.ResponseWriter, r *http.Request) {
t, ok := s.developerTask(w, r, r.PathValue("id"))
if !ok {
return
}
me := CurrentUser(r.Context())
if !t.IsAssignedTo(me.ID) {
writeError(w, http.StatusForbidden, "not_assignee", "only the assignee can start work")
return
}
// assigned → in_progress, and changes_requested → in_progress (resume)
if err := s.store.TransitionTask(r.Context(), t, domain.StatusInProgress,
me.ID, "work_started", nil, nil); err != nil {
s.taskTransitionError(w, err)
return
}
s.Publish("board", "task.updated", map[string]any{"taskId": t.ID, "customerId": t.CustomerID})
w.WriteHeader(http.StatusNoContent)
}
func (s *Server) handleSubmitReview(w http.ResponseWriter, r *http.Request) {
t, ok := s.developerTask(w, r, r.PathValue("id"))
if !ok {
return
}
me := CurrentUser(r.Context())
if !t.IsAssignedTo(me.ID) {
writeError(w, http.StatusForbidden, "not_assignee", "only the assignee can submit for review")
return
}
if err := s.store.TransitionTask(r.Context(), t, domain.StatusInReview,
me.ID, "submitted_for_review", nil, nil); err != nil {
s.taskTransitionError(w, err)
return
}
s.notifyUser(r.Context(), t.ConsultantID, "review_requested",
"Ready for review: "+t.Title, me.Name+" submitted this task.", "/tasks/"+t.ID)
s.Publish("board", "task.updated", map[string]any{"taskId": t.ID, "customerId": t.CustomerID})
w.WriteHeader(http.StatusNoContent)
}
func (s *Server) handleAbandonTask(w http.ResponseWriter, r *http.Request) {
t, ok := s.developerTask(w, r, r.PathValue("id"))
if !ok {
return
}
me := CurrentUser(r.Context())
if !t.IsAssignedTo(me.ID) {
writeError(w, http.StatusForbidden, "not_assignee", "only the assignee can abandon the task")
return
}
if err := s.store.TransitionTask(r.Context(), t, domain.StatusPublished,
me.ID, "abandoned", nil, map[string]any{"assignee": nil}); err != nil {
s.taskTransitionError(w, err)
return
}
s.notifyUser(r.Context(), t.ConsultantID, "task_abandoned",
"Task abandoned: "+t.Title, me.Name+" returned this task to the board.", "/tasks/"+t.ID)
s.Publish("board", "task.published", map[string]any{"taskId": t.ID, "customerId": t.CustomerID})
w.WriteHeader(http.StatusNoContent)
}
var mentionRe = regexp.MustCompile(`@([\w.+-]+@[\w.-]+\.\w+|\w+)`)
func (s *Server) handleAddComment(w http.ResponseWriter, r *http.Request) {
var req struct {
Body string `json:"body"`
}
if !decodeJSON(w, r, &req) {
return
}
body := chat.SanitizeHTML(req.Body)
if strings.TrimSpace(chat.SanitizePlain(body)) == "" {
writeError(w, http.StatusBadRequest, "empty_comment", "comment cannot be empty")
return
}
me := CurrentUser(r.Context())
t, err := s.store.TaskByID(r.Context(), r.PathValue("id"))
if err != nil {
writeError(w, http.StatusNotFound, "not_found", "task not found")
return
}
// commenters: consultants of the customer, the assignee, claimers
canComment := me.Roles.Admin || t.IsAssignedTo(me.ID) || t.HasClaimFrom(me.ID)
if !canComment && me.Roles.Consultant {
if c, err := s.store.CustomerByID(r.Context(), t.CustomerID); err == nil && c.HasConsultant(me.ID) {
canComment = true
}
}
if !canComment {
writeError(w, http.StatusForbidden, "forbidden", "no access to comment on this task")
return
}
comment := domain.Comment{ID: ulid.New(), AuthorID: me.ID, Body: body, At: time.Now().UTC()}
_, err = s.store.DB.Collection("tasks").UpdateOne(r.Context(), bson.M{"_id": t.ID}, bson.M{
"$push": bson.M{"comments": comment},
"$set": bson.M{"updatedAt": time.Now().UTC()},
"$inc": bson.M{"version": 1},
})
if err != nil {
s.internalError(w, r, "add comment", err)
return
}
s.notifyMentions(r, t, me, body)
// always tell the "other side"
if t.IsAssignedTo(me.ID) || t.HasClaimFrom(me.ID) {
s.notifyUser(r.Context(), t.ConsultantID, "comment",
"New comment on "+t.Title, me.Name+" commented.", "/tasks/"+t.ID)
} else if t.Assignee != nil && t.Assignee.Kind == "human" && t.Assignee.UserID != me.ID {
s.notifyUser(r.Context(), t.Assignee.UserID, "comment",
"New comment on "+t.Title, me.Name+" commented.", "/tasks/"+t.ID)
}
s.Publish("board", "task.comment", map[string]any{"taskId": t.ID})
writeJSON(w, http.StatusCreated, map[string]any{"comment": comment})
}
// notifyMentions resolves @name / @email tokens against task participants
// (§11.3).
func (s *Server) notifyMentions(r *http.Request, t *domain.Task, author *domain.User, body string) {
plain := chat.SanitizePlain(body)
matches := mentionRe.FindAllStringSubmatch(plain, 10)
if len(matches) == 0 {
return
}
// candidate participants
ids := map[string]bool{t.ConsultantID: true}
if t.Assignee != nil && t.Assignee.UserID != "" {
ids[t.Assignee.UserID] = true
}
for _, cr := range t.ClaimRequests {
ids[cr.DeveloperID] = true
}
for _, c := range t.Comments {
ids[c.AuthorID] = true
}
notified := map[string]bool{}
for _, m := range matches {
token := strings.ToLower(m[1])
for id := range ids {
if id == author.ID || notified[id] {
continue
}
u, err := s.store.UserByID(r.Context(), id)
if err != nil {
continue
}
first := strings.ToLower(strings.SplitN(u.Name, " ", 2)[0])
if token == strings.ToLower(u.Email) || token == first {
notified[id] = true
s.notifyUser(r.Context(), id, "mention",
author.Name+" mentioned you", "On task: "+t.Title, "/tasks/"+t.ID)
}
}
}
}
func (s *Server) handleLogTime(w http.ResponseWriter, r *http.Request) {
var req struct {
Minutes int `json:"minutes"`
Note string `json:"note"`
}
if !decodeJSON(w, r, &req) {
return
}
if req.Minutes <= 0 || req.Minutes > 24*60 {
writeError(w, http.StatusBadRequest, "bad_minutes", "minutes must be 1..1440")
return
}
t, ok := s.developerTask(w, r, r.PathValue("id"))
if !ok {
return
}
me := CurrentUser(r.Context())
if !t.IsAssignedTo(me.ID) {
writeError(w, http.StatusForbidden, "not_assignee", "only the assignee can log time")
return
}
entry := domain.TimeLogEntry{DeveloperID: me.ID, Minutes: req.Minutes,
Note: req.Note, At: time.Now().UTC()}
_, err := s.store.DB.Collection("tasks").UpdateOne(r.Context(), bson.M{"_id": t.ID}, bson.M{
"$push": bson.M{"timeLog": entry},
"$set": bson.M{"updatedAt": time.Now().UTC()},
"$inc": bson.M{"version": 1},
})
if err != nil {
s.internalError(w, r, "log time", err)
return
}
writeJSON(w, http.StatusCreated, map[string]any{"entry": entry})
}
func (s *Server) handleListNotifications(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query()
me := CurrentUser(r.Context()).ID
list, err := s.store.ListNotifications(r.Context(), me,
q.Get("unread") == "true", q.Get("cursor"), atoiDefault(q.Get("limit"), 30))
if err != nil {
s.internalError(w, r, "list notifications", err)
return
}
unread, err := s.store.CountUnreadNotifications(r.Context(), me)
if err != nil {
s.internalError(w, r, "count unread", err)
return
}
writeJSON(w, http.StatusOK, map[string]any{"notifications": list, "unread": unread})
}
func (s *Server) handleReadNotifications(w http.ResponseWriter, r *http.Request) {
var req struct {
IDs []string `json:"ids"` // empty = mark all read
}
if !decodeJSON(w, r, &req) {
return
}
n, err := s.store.MarkNotificationsRead(r.Context(), CurrentUser(r.Context()).ID, req.IDs)
if err != nil {
s.internalError(w, r, "mark read", err)
return
}
writeJSON(w, http.StatusOK, map[string]any{"marked": n})
}
// handleUserCard serves the public profile hover card (§11.13).
func (s *Server) handleUserCard(w http.ResponseWriter, r *http.Request) {
u, err := s.store.UserByID(r.Context(), r.PathValue("id"))
if err != nil {
writeError(w, http.StatusNotFound, "not_found", "user not found")
return
}
writeJSON(w, http.StatusOK, map[string]any{"card": map[string]any{
"id": u.ID, "name": u.Name, "bio": u.Bio, "roles": u.Roles,
"avatarFileId": u.AvatarFileID, "contact": u.Contact, "extra": u.Extra,
}})
}
// notifyUser inserts an in-app notification and pushes it live.
func (s *Server) notifyUser(ctx context.Context, userID, kind, title, body, link string) {
n := &store.Notification{UserID: userID, Kind: kind, Title: title, Body: body, Link: link}
if err := s.store.InsertNotification(ctx, n); err != nil {
s.log.Warn("insert notification", "err", err)
return
}
s.Publish("notifications", "notification", n)
}
+358
View File
@@ -0,0 +1,358 @@
package httpx
import (
"net/http"
"strconv"
"strings"
"time"
"go.mongodb.org/mongo-driver/v2/bson"
"bountyboard/internal/chat"
"bountyboard/internal/domain"
"bountyboard/internal/files"
"bountyboard/internal/store"
"bountyboard/internal/workperform"
)
func (s *Server) routesConsultant(mux *http.ServeMux) {
con := func(h http.HandlerFunc) http.Handler { return s.authedRole("consultant", h) }
mux.Handle("POST /api/v1/tasks/{id}/approve-claim", con(s.handleApproveClaim))
mux.Handle("POST /api/v1/tasks/{id}/decline-claim", con(s.handleDeclineClaim))
mux.Handle("POST /api/v1/tasks/{id}/unassign", con(s.handleUnassign))
mux.Handle("POST /api/v1/tasks/{id}/assign-ai", con(s.handleAssignAI))
mux.Handle("POST /api/v1/tasks/{id}/review", con(s.handleReview))
mux.Handle("GET /api/v1/consultant/reviews", con(s.handleReviewQueue))
mux.Handle("GET /api/v1/consultant/pool", con(s.handleListPool))
mux.Handle("POST /api/v1/consultant/pool", con(s.handleAddPool))
mux.Handle("DELETE /api/v1/consultant/pool/{developerId}", con(s.handleRemovePool))
}
func (s *Server) handleApproveClaim(w http.ResponseWriter, r *http.Request) {
var req struct {
DeveloperID string `json:"developerId"`
}
if !decodeJSON(w, r, &req) {
return
}
t, _, ok := s.consultantTask(w, r, r.PathValue("id"))
if !ok {
return
}
if !t.HasClaimFrom(req.DeveloperID) {
writeError(w, http.StatusBadRequest, "no_claim", "that developer has not requested this task")
return
}
me := CurrentUser(r.Context())
declined := []string{}
for _, cr := range t.ClaimRequests {
if cr.DeveloperID != req.DeveloperID {
declined = append(declined, cr.DeveloperID)
}
}
err := s.store.TransitionTask(r.Context(), t, domain.StatusAssigned, me.ID, "claim_approved",
map[string]any{"developerId": req.DeveloperID},
map[string]any{
"assignee": domain.Assignee{Kind: "human", UserID: req.DeveloperID},
"claimRequests": []domain.ClaimRequest{},
})
if err != nil {
s.taskTransitionError(w, err)
return
}
s.notifyUser(r.Context(), req.DeveloperID, "claim_approved",
"You got the task: "+t.Title, "Bounty "+formatBounty(t.Bounty), "/tasks/"+t.ID)
for _, id := range declined { // §4.4: declined devs are notified
s.notifyUser(r.Context(), id, "claim_declined",
"Task went to someone else: "+t.Title, "", "/board")
}
s.Publish("board", "task.assigned", map[string]any{"taskId": t.ID, "customerId": t.CustomerID})
w.WriteHeader(http.StatusNoContent)
}
func (s *Server) handleDeclineClaim(w http.ResponseWriter, r *http.Request) {
var req struct {
DeveloperID string `json:"developerId"`
}
if !decodeJSON(w, r, &req) {
return
}
t, _, ok := s.consultantTask(w, r, r.PathValue("id"))
if !ok {
return
}
if !t.HasClaimFrom(req.DeveloperID) {
writeError(w, http.StatusBadRequest, "no_claim", "that developer has not requested this task")
return
}
s.notifyUser(r.Context(), req.DeveloperID, "claim_declined",
"Assignment request declined: "+t.Title, "", "/board")
s.removeClaim(w, r, t, req.DeveloperID, CurrentUser(r.Context()).ID, "claim_declined")
}
func (s *Server) handleUnassign(w http.ResponseWriter, r *http.Request) {
t, _, ok := s.consultantTask(w, r, r.PathValue("id"))
if !ok {
return
}
prev := t.Assignee
// best-effort cancel for AI jobs
if prev != nil && prev.Kind == "ai" && prev.JobID != "" && s.performerClient != nil {
if err := s.performerClient.Cancel(r.Context(), prev.JobID); err != nil {
s.log.Warn("cancel wp job", "jobId", prev.JobID, "err", err)
}
}
err := s.store.TransitionTask(r.Context(), t, domain.StatusPublished,
CurrentUser(r.Context()).ID, "unassigned", nil, map[string]any{"assignee": nil})
if err != nil {
s.taskTransitionError(w, err)
return
}
if prev != nil && prev.Kind == "human" && prev.UserID != "" {
s.notifyUser(r.Context(), prev.UserID, "unassigned",
"You were unassigned: "+t.Title, "The task is back on the board.", "/board")
}
s.Publish("board", "task.published", map[string]any{"taskId": t.ID, "customerId": t.CustomerID})
w.WriteHeader(http.StatusNoContent)
}
func (s *Server) handleAssignAI(w http.ResponseWriter, r *http.Request) {
var req struct {
Context workperform.JobContext `json:"context"`
}
if !decodeJSON(w, r, &req) {
return
}
t, _, ok := s.consultantTask(w, r, r.PathValue("id"))
if !ok {
return
}
if t.Status != domain.StatusPublished {
writeError(w, http.StatusConflict, "bad_status", "only published tasks can be assigned to AI")
return
}
if s.performerClient == nil {
writeError(w, http.StatusServiceUnavailable, "performer_unavailable", "work performer is not configured")
return
}
atts := make([]workperform.Attachment, 0, len(t.Attachments))
for _, a := range t.Attachments {
url := a.URL
if a.FileID != "" {
url = files.SignedURL(s.cfg.AppBaseURL, []byte(s.cfg.SessionSecret), a.FileID,
time.Now().Add(files.DefaultTokenTTL))
}
atts = append(atts, workperform.Attachment{Name: a.Name, URL: url, MimeType: a.MimeType})
}
job, err := s.performerClient.Submit(r.Context(), workperform.JobRequest{
TaskID: t.ID,
Title: t.Title,
Description: t.Description,
AcceptanceCriteria: t.AcceptanceCriteria,
Attachments: atts,
Links: t.Links,
Context: req.Context,
CallbackURL: s.cfg.AppBaseURL + "/api/v1/internal/work-results",
})
if err != nil {
s.log.Error("submit wp job", "task", t.ID, "err", err)
writeError(w, http.StatusBadGateway, "performer_error", "work performer rejected the job: "+err.Error())
return
}
err = s.store.TransitionTask(r.Context(), t, domain.StatusAssigned,
CurrentUser(r.Context()).ID, "assigned_to_ai",
map[string]any{"jobId": job.JobID},
map[string]any{"assignee": domain.Assignee{Kind: "ai", JobID: job.JobID}})
if err != nil {
// roll the job back as best we can
if cErr := s.performerClient.Cancel(r.Context(), job.JobID); cErr != nil {
s.log.Warn("cancel orphaned wp job", "jobId", job.JobID, "err", cErr)
}
s.taskTransitionError(w, err)
return
}
s.Publish("board", "task.assigned", map[string]any{"taskId": t.ID, "customerId": t.CustomerID})
writeJSON(w, http.StatusAccepted, map[string]string{"jobId": job.JobID})
}
func (s *Server) handleReviewQueue(w http.ResponseWriter, r *http.Request) {
u := CurrentUser(r.Context())
consultantID := u.ID
if u.Roles.Admin {
consultantID = ""
}
customers, err := s.store.ListCustomers(r.Context(), consultantID, false)
if err != nil {
s.internalError(w, r, "list customers", err)
return
}
ids := make([]string, len(customers))
for i, c := range customers {
ids[i] = c.ID
}
tasks := []domain.Task{}
if len(ids) > 0 {
tasks, err = s.store.ListTasks(r.Context(), store.TaskFilter{
CustomerIDs: ids,
Statuses: []domain.TaskStatus{domain.StatusInReview},
Sort: "updated",
Limit: 200,
})
if err != nil {
s.internalError(w, r, "list reviews", err)
return
}
}
writeJSON(w, http.StatusOK, map[string]any{"tasks": tasks})
}
func (s *Server) handleReview(w http.ResponseWriter, r *http.Request) {
var req struct {
Decision string `json:"decision"` // approve | request_changes
Note string `json:"note"`
Checklist []struct {
Criterion string `json:"criterion"`
OK bool `json:"ok"`
} `json:"checklist"`
}
if !decodeJSON(w, r, &req) {
return
}
t, _, ok := s.consultantTask(w, r, r.PathValue("id"))
if !ok {
return
}
me := CurrentUser(r.Context())
note := chat.SanitizePlain(req.Note)
// §11.18: per-AC verdicts land on the timeline as the QA audit trail
checklist := make([]map[string]any, 0, len(req.Checklist))
for _, c := range req.Checklist {
checklist = append(checklist, map[string]any{"criterion": c.Criterion, "ok": c.OK})
}
data := map[string]any{"note": note, "checklist": checklist}
switch req.Decision {
case "approve":
err := s.store.TransitionTask(r.Context(), t, domain.StatusApproved, me.ID, "approved", data, nil)
if err != nil {
s.taskTransitionError(w, err)
return
}
// §4.5: immutable award, human assignees only (AI earns no bounty)
if t.Assignee != nil && t.Assignee.Kind == "human" && t.Assignee.UserID != "" {
award := &store.BountyAward{
TaskID: t.ID,
DeveloperID: t.Assignee.UserID,
ConsultantID: t.ConsultantID,
CustomerID: t.CustomerID,
Amount: t.Bounty,
Coefficient: t.EffortCoefficient,
}
if err := s.store.InsertAward(r.Context(), award); err != nil {
s.log.Error("insert bounty award", "task", t.ID, "err", err)
}
s.notifyUser(r.Context(), t.Assignee.UserID, "approved",
"Approved! Bounty awarded: "+formatBounty(t.Bounty), t.Title, "/tasks/"+t.ID)
}
s.metrics.Inc("tasks_approved_total", 1)
case "request_changes":
if strings.TrimSpace(note) == "" {
writeError(w, http.StatusBadRequest, "note_required", "explain what needs to change")
return
}
err := s.store.TransitionTask(r.Context(), t, domain.StatusChangesRequested, me.ID, "changes_requested", data, nil)
if err != nil {
s.taskTransitionError(w, err)
return
}
if t.Assignee != nil && t.Assignee.Kind == "human" && t.Assignee.UserID != "" {
s.notifyUser(r.Context(), t.Assignee.UserID, "changes_requested",
"Changes requested: "+t.Title, note, "/tasks/"+t.ID)
}
s.metrics.Inc("tasks_changes_requested_total", 1)
default:
writeError(w, http.StatusBadRequest, "bad_decision", "decision must be approve or request_changes")
return
}
s.Publish("board", "task.reviewed", map[string]any{"taskId": t.ID, "customerId": t.CustomerID})
w.WriteHeader(http.StatusNoContent)
}
func (s *Server) handleListPool(w http.ResponseWriter, r *http.Request) {
me := CurrentUser(r.Context())
poolIDs, err := s.store.PoolDeveloperIDs(r.Context(), me.ID)
if err != nil {
s.internalError(w, r, "list pool", err)
return
}
inPool := map[string]bool{}
for _, id := range poolIDs {
inPool[id] = true
}
// global developer pool (§3): all enabled developers
cur, err := s.store.DB.Collection("users").Find(r.Context(),
bson.M{"roles.developer": true, "disabled": false})
if err != nil {
s.internalError(w, r, "list developers", err)
return
}
var devs []domain.User
if err := cur.All(r.Context(), &devs); err != nil {
s.internalError(w, r, "decode developers", err)
return
}
out := make([]map[string]any, 0, len(devs))
for _, d := range devs {
out = append(out, map[string]any{
"id": d.ID, "name": d.Name, "email": d.Email, "bio": d.Bio,
"avatarFileId": d.AvatarFileID, "inPool": inPool[d.ID],
})
}
writeJSON(w, http.StatusOK, map[string]any{"developers": out})
}
func (s *Server) handleAddPool(w http.ResponseWriter, r *http.Request) {
var req struct {
DeveloperID string `json:"developerId"`
}
if !decodeJSON(w, r, &req) {
return
}
dev, err := s.store.UserByID(r.Context(), req.DeveloperID)
if err != nil || !dev.Roles.Developer {
writeError(w, http.StatusBadRequest, "not_developer", "that user is not a developer")
return
}
me := CurrentUser(r.Context())
if err := s.store.AddToPool(r.Context(), me.ID, req.DeveloperID); err != nil {
if err == store.ErrDuplicate {
writeError(w, http.StatusConflict, "already_in_pool", "developer is already in your pool")
return
}
s.internalError(w, r, "add to pool", err)
return
}
s.notifyUser(r.Context(), req.DeveloperID, "pool_added",
me.Name+" added you to their developer pool",
"Their published tasks now appear on your bounty board.", "/board")
w.WriteHeader(http.StatusCreated)
}
func (s *Server) handleRemovePool(w http.ResponseWriter, r *http.Request) {
me := CurrentUser(r.Context())
if err := s.store.RemoveFromPool(r.Context(), me.ID, r.PathValue("developerId")); err != nil {
if err == store.ErrNotFound {
writeError(w, http.StatusNotFound, "not_found", "developer is not in your pool")
return
}
s.internalError(w, r, "remove from pool", err)
return
}
w.WriteHeader(http.StatusNoContent)
}
func formatBounty(v float64) string {
return strconv.FormatFloat(v, 'f', -1, 64)
}
+445
View File
@@ -0,0 +1,445 @@
//go:build integration
package httpx
import (
"bytes"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
"go.mongodb.org/mongo-driver/v2/bson"
"bountyboard/internal/domain"
"bountyboard/internal/store"
"bountyboard/internal/workperform"
)
// lifecycleStack: admin-created customer, consultant, developer in pool, and
// one atomized task ready to publish.
type lifecycleStack struct {
ts *httptest.Server
srv *Server
st *store.Store
consultant *http.Client
consCSRF string
consUser domain.User
developer *http.Client
devCSRF string
devUser domain.User
task *domain.Task
customerID string
}
func newLifecycleStack(t *testing.T, extraEnv map[string]string) *lifecycleStack {
t.Helper()
ts, srv, st, _ := newAuthStack(t, extraEnv)
cc := newClient(t)
consUser := registerUser(t, ts.URL, cc, "lc-cons@example.com", "Lifecycle Cons")
promote(t, st, consUser.ID, map[string]bool{"consultant": true})
dc := newClient(t)
devUser := registerUser(t, ts.URL, dc, "lc-dev@example.com", "Lifecycle Dev")
customer := &domain.Customer{
Name: "LC Customer",
Ticketing: domain.Ticketing{Type: domain.TicketingDemo, ProjectKey: "LC"},
ConsultantIDs: []string{consUser.ID},
DefaultBudget: 1000,
}
if err := st.CreateCustomer(t.Context(), customer); err != nil {
t.Fatal(err)
}
// consultant pools the developer
consCSRF := csrfFrom(t, cc, ts.URL)
resp := postJSON(t, cc, ts.URL+"/api/v1/consultant/pool",
map[string]string{"developerId": devUser.ID}, consCSRF)
if resp.StatusCode != http.StatusCreated {
t.Fatalf("add to pool: %d", resp.StatusCode)
}
resp.Body.Close()
task := &domain.Task{
CustomerID: customer.ID, ConsultantID: consUser.ID,
Origin: domain.OriginSubdivided, Status: domain.StatusAtomized,
Title: "Build the CSV importer", Description: "as specified",
AcceptanceCriteria: []string{"imports valid rows", "reports bad rows"},
EffortCoefficient: 0.25, Budget: 1000, Bounty: 250,
}
if err := st.InsertTask(t.Context(), task); err != nil {
t.Fatal(err)
}
return &lifecycleStack{
ts: ts, srv: srv, st: st,
consultant: cc, consCSRF: consCSRF, consUser: consUser,
developer: dc, devCSRF: csrfFrom(t, dc, ts.URL), devUser: devUser,
task: task, customerID: customer.ID,
}
}
func (l *lifecycleStack) taskStatus(t *testing.T) domain.TaskStatus {
t.Helper()
fresh, err := l.st.TaskByID(t.Context(), l.task.ID)
if err != nil {
t.Fatal(err)
}
return fresh.Status
}
func mustPost(t *testing.T, c *http.Client, url string, body any, csrf string, want int) *http.Response {
t.Helper()
resp := postJSON(t, c, url, body, csrf)
if resp.StatusCode != want {
var buf bytes.Buffer
buf.ReadFrom(resp.Body)
t.Fatalf("POST %s: %d (want %d): %s", url, resp.StatusCode, want, buf.String())
}
return resp
}
func TestFullLifecycleImportedToApproved(t *testing.T) {
l := newLifecycleStack(t, nil)
base := l.ts.URL + "/api/v1/tasks/" + l.task.ID
// consultant publishes
mustPost(t, l.consultant, base+"/publish", map[string]any{}, l.consCSRF, http.StatusOK).Body.Close()
if got := l.taskStatus(t); got != domain.StatusPublished {
t.Fatalf("after publish: %s", got)
}
// developer sees it on the board with the right bounty
resp, err := l.developer.Get(l.ts.URL + "/api/v1/board")
if err != nil {
t.Fatal(err)
}
var board struct {
Tasks []domain.Task `json:"tasks"`
}
bodyJSON(t, resp, &board)
if len(board.Tasks) != 1 || board.Tasks[0].Bounty != 250 {
t.Fatalf("board: %+v", board.Tasks)
}
// developer claims
mustPost(t, l.developer, base+"/claim", map[string]string{"note": "I know CSV"}, l.devCSRF, http.StatusNoContent).Body.Close()
if got := l.taskStatus(t); got != domain.StatusClaimRequested {
t.Fatalf("after claim: %s", got)
}
// premature lifecycle calls are rejected
resp = postJSON(t, l.developer, base+"/start", map[string]any{}, l.devCSRF)
if resp.StatusCode != http.StatusForbidden { // not assignee yet
t.Fatalf("start before assignment: %d", resp.StatusCode)
}
resp.Body.Close()
// consultant approves the claim
mustPost(t, l.consultant, base+"/approve-claim",
map[string]string{"developerId": l.devUser.ID}, l.consCSRF, http.StatusNoContent).Body.Close()
if got := l.taskStatus(t); got != domain.StatusAssigned {
t.Fatalf("after approve-claim: %s", got)
}
// developer starts, logs time, comments, submits
mustPost(t, l.developer, base+"/start", map[string]any{}, l.devCSRF, http.StatusNoContent).Body.Close()
mustPost(t, l.developer, base+"/time",
map[string]any{"minutes": 90, "note": "parser"}, l.devCSRF, http.StatusCreated).Body.Close()
mustPost(t, l.developer, base+"/comments",
map[string]any{"body": "<p>Done, see <b>PR</b> <script>alert(1)</script></p>"},
l.devCSRF, http.StatusCreated).Body.Close()
mustPost(t, l.developer, base+"/submit-review", map[string]any{}, l.devCSRF, http.StatusNoContent).Body.Close()
if got := l.taskStatus(t); got != domain.StatusInReview {
t.Fatalf("after submit: %s", got)
}
// comment was sanitized
fresh, _ := l.st.TaskByID(t.Context(), l.task.ID)
if len(fresh.Comments) != 1 || bytes.Contains([]byte(fresh.Comments[0].Body), []byte("<script")) {
t.Fatalf("comment sanitization: %+v", fresh.Comments)
}
if len(fresh.TimeLog) != 1 || fresh.TimeLog[0].Minutes != 90 {
t.Fatalf("time log: %+v", fresh.TimeLog)
}
// consultant reviews in the queue, then requests changes
qResp, err := l.consultant.Get(l.ts.URL + "/api/v1/consultant/reviews")
if err != nil {
t.Fatal(err)
}
var queue struct {
Tasks []domain.Task `json:"tasks"`
}
bodyJSON(t, qResp, &queue)
if len(queue.Tasks) != 1 {
t.Fatalf("review queue: %d tasks", len(queue.Tasks))
}
mustPost(t, l.consultant, base+"/review",
map[string]any{"decision": "request_changes", "note": "handle BOM"},
l.consCSRF, http.StatusNoContent).Body.Close()
if got := l.taskStatus(t); got != domain.StatusChangesRequested {
t.Fatalf("after request_changes: %s", got)
}
// developer resumes and resubmits; consultant approves with checklist
mustPost(t, l.developer, base+"/start", map[string]any{}, l.devCSRF, http.StatusNoContent).Body.Close()
mustPost(t, l.developer, base+"/submit-review", map[string]any{}, l.devCSRF, http.StatusNoContent).Body.Close()
mustPost(t, l.consultant, base+"/review", map[string]any{
"decision": "approve", "note": "nice",
"checklist": []map[string]any{
{"criterion": "imports valid rows", "ok": true},
{"criterion": "reports bad rows", "ok": true},
},
}, l.consCSRF, http.StatusNoContent).Body.Close()
if got := l.taskStatus(t); got != domain.StatusApproved {
t.Fatalf("after approve: %s", got)
}
// immutable bountyAwards row exists with the frozen amount
var award store.BountyAward
if err := l.st.DB.Collection("bountyAwards").FindOne(t.Context(),
bson.M{"taskId": l.task.ID}).Decode(&award); err != nil {
t.Fatalf("award row: %v", err)
}
if award.DeveloperID != l.devUser.ID || award.Amount != 250 || award.Coefficient != 0.25 {
t.Fatalf("award: %+v", award)
}
// the review checklist landed on the timeline (§11.18)
fresh, _ = l.st.TaskByID(t.Context(), l.task.ID)
var checklistOK bool
for _, e := range fresh.Timeline {
if e.Event == "approved" && e.Data["checklist"] != nil {
checklistOK = true
}
}
if !checklistOK {
t.Error("timeline missing approval checklist")
}
// developer got the approval notification
notifs, _ := l.st.ListNotifications(t.Context(), l.devUser.ID, false, "", 50)
var approvedNotif bool
for _, n := range notifs {
if n.Kind == "approved" {
approvedNotif = true
}
}
if !approvedNotif {
t.Error("developer missing approval notification")
}
// approved task is immutable
resp = patchJSON(t, l.consultant, base, map[string]any{"budget": 9999.0}, l.consCSRF)
if resp.StatusCode != http.StatusConflict {
t.Fatalf("edit after approve: %d, want 409", resp.StatusCode)
}
resp.Body.Close()
}
func TestDeclineWithdrawUnassignAbandon(t *testing.T) {
l := newLifecycleStack(t, nil)
base := l.ts.URL + "/api/v1/tasks/" + l.task.ID
mustPost(t, l.consultant, base+"/publish", map[string]any{}, l.consCSRF, http.StatusOK).Body.Close()
// claim + consultant declines → back to published, dev notified
mustPost(t, l.developer, base+"/claim", map[string]string{"note": ""}, l.devCSRF, http.StatusNoContent).Body.Close()
mustPost(t, l.consultant, base+"/decline-claim",
map[string]string{"developerId": l.devUser.ID}, l.consCSRF, http.StatusNoContent).Body.Close()
if got := l.taskStatus(t); got != domain.StatusPublished {
t.Fatalf("after decline: %s", got)
}
notifs, _ := l.st.ListNotifications(t.Context(), l.devUser.ID, false, "", 50)
var declined bool
for _, n := range notifs {
if n.Kind == "claim_declined" {
declined = true
}
}
if !declined {
t.Error("declined developer was not notified")
}
// claim + withdraw → back to published
mustPost(t, l.developer, base+"/claim", map[string]string{}, l.devCSRF, http.StatusNoContent).Body.Close()
mustPost(t, l.developer, base+"/claim/withdraw", map[string]any{}, l.devCSRF, http.StatusNoContent).Body.Close()
if got := l.taskStatus(t); got != domain.StatusPublished {
t.Fatalf("after withdraw: %s", got)
}
// assign → unassign by consultant → published again
mustPost(t, l.developer, base+"/claim", map[string]string{}, l.devCSRF, http.StatusNoContent).Body.Close()
mustPost(t, l.consultant, base+"/approve-claim",
map[string]string{"developerId": l.devUser.ID}, l.consCSRF, http.StatusNoContent).Body.Close()
mustPost(t, l.consultant, base+"/unassign", map[string]any{}, l.consCSRF, http.StatusNoContent).Body.Close()
if got := l.taskStatus(t); got != domain.StatusPublished {
t.Fatalf("after unassign: %s", got)
}
fresh, _ := l.st.TaskByID(t.Context(), l.task.ID)
if fresh.Assignee != nil {
t.Fatalf("assignee not cleared: %+v", fresh.Assignee)
}
// assign again → developer abandons mid-progress → published
mustPost(t, l.developer, base+"/claim", map[string]string{}, l.devCSRF, http.StatusNoContent).Body.Close()
mustPost(t, l.consultant, base+"/approve-claim",
map[string]string{"developerId": l.devUser.ID}, l.consCSRF, http.StatusNoContent).Body.Close()
mustPost(t, l.developer, base+"/start", map[string]any{}, l.devCSRF, http.StatusNoContent).Body.Close()
mustPost(t, l.developer, base+"/abandon", map[string]any{}, l.devCSRF, http.StatusNoContent).Body.Close()
if got := l.taskStatus(t); got != domain.StatusPublished {
t.Fatalf("after abandon: %s", got)
}
}
func TestDeveloperOutsidePoolSeesNothing(t *testing.T) {
l := newLifecycleStack(t, nil)
base := l.ts.URL + "/api/v1/tasks/" + l.task.ID
mustPost(t, l.consultant, base+"/publish", map[string]any{}, l.consCSRF, http.StatusOK).Body.Close()
stranger := newClient(t)
registerUser(t, l.ts.URL, stranger, "stranger@example.com", "Stranger")
strangerCSRF := csrfFrom(t, stranger, l.ts.URL)
resp, err := stranger.Get(l.ts.URL + "/api/v1/board")
if err != nil {
t.Fatal(err)
}
var board struct {
Tasks []domain.Task `json:"tasks"`
}
bodyJSON(t, resp, &board)
if len(board.Tasks) != 0 {
t.Fatalf("stranger sees %d tasks", len(board.Tasks))
}
resp = postJSON(t, stranger, base+"/claim", map[string]string{}, strangerCSRF)
if resp.StatusCode != http.StatusForbidden {
t.Fatalf("stranger claim: %d", resp.StatusCode)
}
resp.Body.Close()
}
func TestAIAssignmentAndCallback(t *testing.T) {
const wpToken = "test-wp-token-abc"
// fake work performer + artifact host
artifact := []byte("diff --git a/x b/x\n+fixed\n")
var fake *httptest.Server
fake = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch {
case r.Method == "POST" && r.URL.Path == "/v1/jobs":
if r.Header.Get("Authorization") != "Bearer "+wpToken {
http.Error(w, "bad token", http.StatusUnauthorized)
return
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusAccepted)
json.NewEncoder(w).Encode(map[string]string{"jobId": "wp_test_1", "status": "queued"})
case r.URL.Path == "/healthz":
w.WriteHeader(http.StatusOK)
case r.URL.Path == "/artifacts/patch.diff":
w.Write(artifact)
default:
http.NotFound(w, r)
}
}))
defer fake.Close()
l := newLifecycleStack(t, map[string]string{"WORK_PERFORMER_TOKEN": wpToken})
l.srv.SetPerformerClient(workperform.New(func() string { return fake.URL }, wpToken, 5*time.Second))
base := l.ts.URL + "/api/v1/tasks/" + l.task.ID
mustPost(t, l.consultant, base+"/publish", map[string]any{}, l.consCSRF, http.StatusOK).Body.Close()
// consultant assigns to AI
resp := mustPost(t, l.consultant, base+"/assign-ai",
map[string]any{"context": map[string]string{"instructions": "be careful"}},
l.consCSRF, http.StatusAccepted)
var accepted struct {
JobID string `json:"jobId"`
}
bodyJSON(t, resp, &accepted)
if accepted.JobID != "wp_test_1" {
t.Fatalf("jobId = %q", accepted.JobID)
}
if got := l.taskStatus(t); got != domain.StatusAssigned {
t.Fatalf("after assign-ai: %s", got)
}
// the service calls back with an HMAC-signed result
cb := workperform.Callback{
JobID: "wp_test_1", TaskID: l.task.ID, Status: "succeeded",
Summary: "Implemented the importer", Log: "ran fine",
}
cb.Artifacts = []struct {
Name string `json:"name"`
URL string `json:"url"`
}{{Name: "patch.diff", URL: fake.URL + "/artifacts/patch.diff"}}
body, _ := json.Marshal(cb)
post := func(sig string) *http.Response {
req, _ := http.NewRequest(http.MethodPost,
l.ts.URL+"/api/v1/internal/work-results", bytes.NewReader(body))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-Signature", sig)
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatal(err)
}
return resp
}
// wrong signature rejected
r1 := post("deadbeef")
if r1.StatusCode != http.StatusUnauthorized {
t.Fatalf("bad signature: %d", r1.StatusCode)
}
r1.Body.Close()
// valid signature moves the task to in_review with ingested artifact
r2 := post(workperform.Sign(wpToken, body))
if r2.StatusCode != http.StatusOK {
t.Fatalf("callback: %d", r2.StatusCode)
}
r2.Body.Close()
if got := l.taskStatus(t); got != domain.StatusInReview {
t.Fatalf("after callback: %s", got)
}
fresh, _ := l.st.TaskByID(t.Context(), l.task.ID)
var stored *domain.TaskAttachment
for i := range fresh.Attachments {
if fresh.Attachments[i].Name == "patch.diff" {
stored = &fresh.Attachments[i]
}
}
if stored == nil || stored.FileID == "" {
t.Fatalf("artifact not ingested: %+v", fresh.Attachments)
}
// duplicate callback is idempotent (no error, no double transition)
r3 := post(workperform.Sign(wpToken, body))
if r3.StatusCode != http.StatusOK {
t.Fatalf("duplicate callback: %d", r3.StatusCode)
}
var dup map[string]string
bodyJSON(t, r3, &dup)
if dup["status"] != "duplicate_ignored" {
t.Fatalf("duplicate callback status: %v", dup)
}
// consultant reviews the AI work exactly like a human's
mustPost(t, l.consultant, base+"/review",
map[string]any{"decision": "approve", "note": "AI did fine"},
l.consCSRF, http.StatusNoContent).Body.Close()
if got := l.taskStatus(t); got != domain.StatusApproved {
t.Fatalf("after AI approve: %s", got)
}
// no bounty award for AI assignees
n, _ := l.st.DB.Collection("bountyAwards").CountDocuments(t.Context(), bson.M{"taskId": l.task.ID})
if n != 0 {
t.Fatalf("AI task produced %d award rows, want 0", n)
}
}
+19 -8
View File
@@ -14,6 +14,7 @@ import (
"bountyboard/internal/files"
"bountyboard/internal/metrics"
"bountyboard/internal/store"
"bountyboard/internal/workperform"
)
// shutdownBudget is the §12 graceful-shutdown drain window.
@@ -33,14 +34,21 @@ type Server struct {
httpSrv *http.Server
// late-bound subsystem hooks (see providers.go)
atomizer BreakerInfo
performer BreakerInfo
syncProvider StatusProvider
jobsProvider StatusProvider
syncTrigger func(customerID string)
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)
atomizer BreakerInfo
performer BreakerInfo
syncProvider StatusProvider
jobsProvider StatusProvider
syncTrigger func(customerID string)
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)
performerClient *workperform.Client
}
// SetPerformerClient wires the §5.2 client (also feeds the status panel).
func (s *Server) SetPerformerClient(c *workperform.Client) {
s.performerClient = c
s.performer = c
}
func New(cfg *config.Config, log *slog.Logger, reg *metrics.Registry, st *store.Store, fs *files.Store) *Server {
@@ -70,6 +78,9 @@ func New(cfg *config.Config, log *slog.Logger, reg *metrics.Registry, st *store.
s.routesProfile(mux)
s.routesAdmin(mux)
s.routesTasks(mux)
s.routesBoard(mux)
s.routesConsultant(mux)
s.routesWorkResults(mux)
mux.HandleFunc("GET /ws", s.handleWS)
s.routesWeb(mux)
+29 -6
View File
@@ -192,14 +192,37 @@ func (s *Server) routesWeb(mux *http.ServeMux) {
})
}))
rolePage := func(path, tmpl, title, active, script string, check func(*domain.User) bool) {
mux.HandleFunc("GET "+path, s.page(func(w http.ResponseWriter, r *http.Request, u *domain.User) {
if check != nil && !check(u) && !u.Roles.Admin {
s.render(w, r, "placeholder.html", &pageData{Title: "Not authorized", User: u})
return
}
s.render(w, r, tmpl, &pageData{
Title: title, Active: active, User: u, Scripts: []string{script},
})
}))
}
isDev := func(u *domain.User) bool { return u.Roles.Developer }
isCons := func(u *domain.User) bool { return u.Roles.Consultant }
rolePage("/board", "board.html", "Bounty Board", "board", "/static/js/board.js", isDev)
rolePage("/my-tasks", "my-tasks.html", "My Tasks", "my-tasks", "/static/js/my-tasks.js", isDev)
rolePage("/consultant/reviews", "reviews.html", "Review Queue", "reviews", "/static/js/reviews.js", isCons)
rolePage("/consultant/pool", "pool.html", "Developer Pool", "pool", "/static/js/pool.js", isCons)
rolePage("/notifications", "notifications.html", "Notifications", "", "/static/js/notifications-page.js", nil)
mux.HandleFunc("GET /tasks/{id}", s.page(func(w http.ResponseWriter, r *http.Request, u *domain.User) {
s.render(w, r, "task.html", &pageData{
Title: "Task", User: u,
Scripts: []string{"/static/js/task.js"},
Data: map[string]any{"TaskID": r.PathValue("id")},
})
}))
// Placeholders for areas built in later phases; replaced as they land.
placeholders := map[string]string{
"/board": "Bounty Board",
"/my-tasks": "My Tasks",
"/consultant/reviews": "Review Queue",
"/consultant/pool": "Developer Pool",
"/messages": "Messages",
"/metrics": "Metrics",
"/messages": "Messages",
"/metrics": "Metrics",
}
for path, title := range placeholders {
mux.HandleFunc("GET "+path, s.page(func(w http.ResponseWriter, r *http.Request, u *domain.User) {
+141
View File
@@ -0,0 +1,141 @@
package httpx
import (
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"time"
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo"
"bountyboard/internal/domain"
"bountyboard/internal/files"
"bountyboard/internal/workperform"
)
func (s *Server) routesWorkResults(mux *http.ServeMux) {
// Service-to-app callback: authenticated by HMAC signature, not session.
mux.HandleFunc("POST /api/v1/internal/work-results", s.handleWorkResults)
}
// handleWorkResults implements the §5.2 callback: signature-verified,
// idempotent by jobId, artifacts ingested into GridFS so review survives the
// performer container.
func (s *Server) handleWorkResults(w http.ResponseWriter, r *http.Request) {
body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 4<<20))
if err != nil {
writeError(w, http.StatusBadRequest, "bad_request", "unreadable body")
return
}
if !workperform.VerifySignature(s.cfg.WorkPerformerToken, body, r.Header.Get("X-Signature")) {
s.metrics.Inc("wp_callback_bad_signature_total", 1)
writeError(w, http.StatusUnauthorized, "bad_signature", "invalid X-Signature")
return
}
var cb workperform.Callback
if err := json.Unmarshal(body, &cb); err != nil {
writeError(w, http.StatusBadRequest, "bad_request", "invalid JSON: "+err.Error())
return
}
if cb.JobID == "" || cb.TaskID == "" {
writeError(w, http.StatusBadRequest, "bad_request", "jobId and taskId are required")
return
}
// Idempotency by jobId: first writer wins, duplicates are acknowledged
// without effect (§5.2).
_, err = s.store.DB.Collection("wpCallbacks").InsertOne(r.Context(), bson.M{
"_id": cb.JobID, "taskId": cb.TaskID, "status": cb.Status, "at": time.Now().UTC(),
})
if err != nil {
if mongo.IsDuplicateKeyError(err) {
writeJSON(w, http.StatusOK, map[string]string{"status": "duplicate_ignored"})
return
}
s.internalError(w, r, "record callback", err)
return
}
t, err := s.store.TaskByID(r.Context(), cb.TaskID)
if err != nil {
writeError(w, http.StatusNotFound, "not_found", "task not found")
return
}
if t.Assignee == nil || t.Assignee.Kind != "ai" || t.Assignee.JobID != cb.JobID {
writeError(w, http.StatusConflict, "job_mismatch", "task is not assigned to this job")
return
}
switch cb.Status {
case "succeeded":
attachments := s.ingestArtifacts(r.Context(), t, cb)
err = s.store.TransitionTask(r.Context(), t, domain.StatusInReview, "system:ai", "ai_completed",
map[string]any{"jobId": cb.JobID, "summary": cb.Summary, "log": tail(cb.Log, 4000)},
map[string]any{"attachments": attachments})
if err != nil {
s.taskTransitionError(w, err)
return
}
s.notifyUser(r.Context(), t.ConsultantID, "review_requested",
"AI finished: "+t.Title, cb.Summary, "/tasks/"+t.ID)
case "failed":
// status stays assigned (§5.2); record + notify
if err := s.store.AppendTimeline(r.Context(), t.ID, domain.TimelineEntry{
ActorID: "system:ai", Event: "ai_failed",
Data: map[string]any{"jobId": cb.JobID, "summary": cb.Summary, "log": tail(cb.Log, 4000)},
}); err != nil {
s.internalError(w, r, "record ai failure", err)
return
}
s.notifyUser(r.Context(), t.ConsultantID, "ai_failed",
"AI work failed: "+t.Title, cb.Summary, "/tasks/"+t.ID)
default:
writeError(w, http.StatusBadRequest, "bad_status", "status must be succeeded or failed")
return
}
s.Publish("board", "task.updated", map[string]any{"taskId": t.ID, "customerId": t.CustomerID})
writeJSON(w, http.StatusOK, map[string]string{"status": "accepted"})
}
// ingestArtifacts downloads every artifact URL into GridFS; failures are
// recorded but never block the review (the summary alone is reviewable).
func (s *Server) ingestArtifacts(ctx context.Context, t *domain.Task, cb workperform.Callback) []domain.TaskAttachment {
attachments := append([]domain.TaskAttachment{}, t.Attachments...)
client := &http.Client{Timeout: 60 * time.Second}
for _, a := range cb.Artifacts {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, a.URL, nil)
if err != nil {
continue
}
resp, err := client.Do(req)
if err != nil {
s.log.Warn("download artifact", "url", a.URL, "err", err)
continue
}
if resp.StatusCode != http.StatusOK {
resp.Body.Close()
s.log.Warn("download artifact", "url", a.URL, "status", resp.StatusCode)
continue
}
meta, err := s.files.Save(ctx, files.ScopeTask, "system:ai", a.Name, "", resp.Body)
resp.Body.Close()
if err != nil {
s.log.Warn("store artifact", "name", a.Name, "err", err)
continue
}
attachments = append(attachments, domain.TaskAttachment{
Name: a.Name, URL: "/files/" + meta.ID, MimeType: meta.MimeType, FileID: meta.ID,
})
}
return attachments
}
func tail(s string, n int) string {
if len(s) <= n {
return s
}
return fmt.Sprintf("…%s", s[len(s)-n:])
}