BountyBoard/internal/http/workresults.go

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:])
}