BountyBoard/internal/atomize/service_integration_test.go

//go:build integration

package atomize

import (
	"context"
	"encoding/json"
	"io"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"os"
	"sync/atomic"
	"testing"
	"time"

	"bountyboard/internal/domain"
	"bountyboard/internal/jobs"
	"bountyboard/internal/metrics"
	"bountyboard/internal/store"
	"bountyboard/internal/testutil"
	"bountyboard/internal/ulid"
)

type stack struct {
	st     *store.Store
	svc    *Service
	runner *jobs.Runner
	task   *domain.Task
}

func newStack(t *testing.T, atomizerHandler http.HandlerFunc) *stack {
	t.Helper()
	db := testutil.MongoDB(t)
	if err := store.EnsureIndexes(t.Context(), db); err != nil {
		t.Fatal(err)
	}
	st := &store.Store{Client: db.Client(), DB: db}

	srv := httptest.NewServer(atomizerHandler)
	t.Cleanup(srv.Close)
	client := New(func() string { return srv.URL }, "tok", 10*time.Second)

	logOut := io.Writer(io.Discard)
	if os.Getenv("TEST_LOG") == "1" {
		logOut = os.Stderr
	}
	log := slog.New(slog.NewTextHandler(logOut, nil))
	svc := NewService(st, client, log, "http://localhost:8787",
		[]byte("0123456789abcdef0123456789abcdef"), nil)
	runner := jobs.NewRunner(st, log, metrics.NewRegistry(), 2)
	runner.Register(JobKindSubdivide, svc.HandleSubdivide)
	runner.Register(JobKindExtend, svc.HandleExtend)

	task := &domain.Task{
		CustomerID: ulid.New(), ConsultantID: ulid.New(),
		Origin: domain.OriginImported, Status: domain.StatusImported,
		Title: "Implement user import", Description: "Full description",
		AcceptanceCriteria: []string{"works"},
		Budget:             1000, EffortCoefficient: 1, Bounty: 1000,
		External: &domain.External{System: "demo", Key: "DEMO-1", URL: "https://demo.invalid/1"},
	}
	if err := st.InsertTask(t.Context(), task); err != nil {
		t.Fatal(err)
	}
	return &stack{st: st, svc: svc, runner: runner, task: task}
}

// runJob synchronously claims + executes pending jobs until idle.
func runJobs(t *testing.T, s *stack, deadline time.Duration) {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), deadline)
	defer cancel()
	go s.runner.Run(ctx)
	<-ctx.Done()
}

func goodAtomizer(calls *atomic.Int64) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		if calls != nil {
			calls.Add(1)
		}
		w.Header().Set("Content-Type", "application/json")
		switch r.URL.Path {
		case "/v1/atomize":
			json.NewEncoder(w).Encode(AtomizeResponse{
				Model: "claude-sonnet-4-6",
				Tasks: []SubTask{
					{Title: "Backend", Description: "API part", AcceptanceCriteria: []string{"a"}, EffortCoefficient: 0.5},
					{Title: "Frontend", Description: "UI part", AcceptanceCriteria: []string{"b"}, EffortCoefficient: 0.3},
					{Title: "Migration", Description: "DB part", AcceptanceCriteria: []string{"c"}, EffortCoefficient: 0.2},
				},
			})
		case "/v1/extend":
			json.NewEncoder(w).Encode(ExtendResponse{
				Task: SubTask{Title: "CSV mapping presets", Description: "extension",
					AcceptanceCriteria: []string{"x"}, EffortCoefficient: 0.4},
			})
		default:
			http.NotFound(w, r)
		}
	}
}

func TestSubdivideEndToEnd(t *testing.T) {
	s := newStack(t, goodAtomizer(nil))
	ctx := context.Background()

	// consultant moved it to atomizing (handler does this before enqueue)
	if err := s.st.TransitionTask(ctx, s.task, domain.StatusAtomizing, "cons", "subdivide_requested", nil, nil); err != nil {
		t.Fatal(err)
	}
	if _, err := s.runner.Enqueue(ctx, JobKindSubdivide, SubdividePayload{
		TaskID: s.task.ID, Note: "split it", ActorID: "cons", FromStatus: "imported",
	}); err != nil {
		t.Fatal(err)
	}
	runJobs(t, s, 4*time.Second)

	parent, err := s.st.TaskByID(ctx, s.task.ID)
	if err != nil {
		t.Fatal(err)
	}
	if parent.Status != domain.StatusAtomized {
		t.Fatalf("parent status = %s, want atomized", parent.Status)
	}
	if parent.AtomizationNote != "split it" {
		t.Errorf("atomizationNote = %q", parent.AtomizationNote)
	}

	children, err := s.st.ListTasks(ctx, store.TaskFilter{ParentID: s.task.ID})
	if err != nil {
		t.Fatal(err)
	}
	if len(children) != 3 {
		t.Fatalf("children = %d, want 3", len(children))
	}
	sum := 0.0
	for _, c := range children {
		if c.Status != domain.StatusAtomized || c.Origin != domain.OriginSubdivided {
			t.Errorf("child %s: status=%s origin=%s", c.ID, c.Status, c.Origin)
		}
		if c.RootID != s.task.ID || c.Budget != 1000 {
			t.Errorf("child %s: root=%s budget=%v", c.ID, c.RootID, c.Budget)
		}
		if c.Bounty != domain.ComputeBounty(c.EffortCoefficient, c.Budget) {
			t.Errorf("child %s: bounty %v != coeff %v × budget %v", c.ID, c.Bounty, c.EffortCoefficient, c.Budget)
		}
		sum += c.EffortCoefficient
	}
	if sum < 0.999 || sum > 1.001 {
		t.Errorf("children coefficients sum %v, want 1±0.001", sum)
	}

	// consultant got the "ready" notification
	notifs, _ := s.st.ListNotifications(ctx, s.task.ConsultantID, false, "", 10)
	found := false
	for _, n := range notifs {
		if n.Kind == "subdivision_ready" {
			found = true
		}
	}
	if !found {
		t.Error("missing subdivision_ready notification")
	}
}

func TestExtendEndToEnd(t *testing.T) {
	s := newStack(t, goodAtomizer(nil))
	ctx := context.Background()

	if _, err := s.runner.Enqueue(ctx, JobKindExtend, ExtendPayload{
		TaskID: s.task.ID, Note: "add CSV presets", ActorID: "cons",
	}); err != nil {
		t.Fatal(err)
	}
	runJobs(t, s, 4*time.Second)

	siblings, err := s.st.ListTasks(ctx, store.TaskFilter{ParentID: s.task.ID})
	if err != nil {
		t.Fatal(err)
	}
	if len(siblings) != 1 {
		t.Fatalf("siblings = %d, want exactly 1", len(siblings))
	}
	sib := siblings[0]
	if sib.Origin != domain.OriginExtended || sib.Status != domain.StatusAtomized {
		t.Fatalf("sibling: %s/%s", sib.Origin, sib.Status)
	}
	if sib.EffortCoefficient != 0.4 || sib.Bounty != 400 {
		t.Fatalf("sibling coeff=%v bounty=%v, want 0.4/400", sib.EffortCoefficient, sib.Bounty)
	}
	// source status unchanged, timeline notes the extension
	src, _ := s.st.TaskByID(ctx, s.task.ID)
	if src.Status != domain.StatusImported {
		t.Fatalf("source status = %s, want imported", src.Status)
	}
}

func TestSubdivideFailureRevertsAfterRetries(t *testing.T) {
	var calls atomic.Int64
	s := newStack(t, func(w http.ResponseWriter, r *http.Request) {
		calls.Add(1)
		http.Error(w, `{"error":{"code":"llm_unavailable","message":"model down"}}`, http.StatusBadGateway)
	})
	ctx := context.Background()
	if err := s.st.TransitionTask(ctx, s.task, domain.StatusAtomizing, "cons", "subdivide_requested", nil, nil); err != nil {
		t.Fatal(err)
	}
	if _, err := s.runner.Enqueue(ctx, JobKindSubdivide, SubdividePayload{
		TaskID: s.task.ID, ActorID: "cons", FromStatus: "imported",
	}); err != nil {
		t.Fatal(err)
	}

	// job retries with backoff (10s, 20s) — run one attempt per window (the
	// HTTP client's own 5xx retries take ~4s), forcing queued retries due
	// immediately between windows to keep the test fast
	runJobs(t, s, 6*time.Second)
	for i := 0; i < 2; i++ {
		if _, err := s.st.DB.Collection("jobs").UpdateMany(ctx,
			map[string]any{"status": jobs.StatusQueued},
			map[string]any{"$set": map[string]any{"runAfter": time.Now().UTC().Add(-time.Second)}}); err != nil {
			t.Fatal(err)
		}
		runJobs(t, s, 6*time.Second)
	}

	parent, err := s.st.TaskByID(ctx, s.task.ID)
	if err != nil {
		t.Fatal(err)
	}
	if parent.Status != domain.StatusImported {
		t.Fatalf("status after exhausted retries = %s, want imported", parent.Status)
	}
	foundFailure := false
	for _, e := range parent.Timeline {
		if e.Event == "atomization_failed" {
			foundFailure = true
		}
	}
	if !foundFailure {
		t.Error("timeline missing atomization_failed")
	}
	notifs, _ := s.st.ListNotifications(ctx, s.task.ConsultantID, false, "", 10)
	failNotif := false
	for _, n := range notifs {
		if n.Kind == "atomization_failed" {
			failNotif = true
		}
	}
	if !failNotif {
		t.Error("missing atomization_failed notification")
	}
}