diff --git a/PROGRESS.md b/PROGRESS.md index d4da1c2..0996631 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -7,3 +7,4 @@ - Phase 4 (base UI shell): embedded templates + static assets (no build step), §10 beige/dark tokens with radius 2px, theme toggle persisted to localStorage + profile, login/register/change-password pages, profile page (avatar upload w/ MIME check, bio, contacts, arbitrary extra fields, version-conflict 409), role-based navigation, /files/{id} serving (session or signed token), CSP without inline scripts + security headers — unit + integration tests green, stack verified live. - Phase 5 (admin): customer CRUD wizard with per-system encrypted credentials + test connection (jira/ado/youtrack/demo connectors), user management (roles, disable w/ session revoke, force-reset, delete, self-lockout guards), runtime settings doc, audit log on every privileged mutation + list API, service-status panel (mongo/atomizer/WP health + latency, late-bound breaker/sync/jobs providers), tabbed admin UI - tests green. Fixed mongod WT_PANIC: container nofile ulimit was 1024, raised to 64000 in compose. - 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). diff --git a/cmd/app/main.go b/cmd/app/main.go index a87c549..3c17fc1 100644 --- a/cmd/app/main.go +++ b/cmd/app/main.go @@ -6,20 +6,25 @@ import ( "encoding/json" "fmt" "log/slog" + "net/http" "os" "os/signal" + "strings" "syscall" "time" + "bountyboard/internal/atomize" "bountyboard/internal/auth" "bountyboard/internal/config" "bountyboard/internal/crypto" "bountyboard/internal/domain" "bountyboard/internal/files" httpx "bountyboard/internal/http" + "bountyboard/internal/jobs" "bountyboard/internal/metrics" "bountyboard/internal/store" syncpkg "bountyboard/internal/sync" + "bountyboard/internal/ws" ) func main() { @@ -69,6 +74,40 @@ func run() error { Probe: st.Ping, }) + // WebSocket hub: one multiplexed live channel per session (§2.2). + hub := ws.NewHub(log, func(r *http.Request) bool { + origin := r.Header.Get("Origin") + if origin == "" { + return true // non-browser client (no CSRF surface on a read feed) + } + return strings.HasPrefix(origin, cfg.AppBaseURL) || + strings.HasPrefix(origin, "http://"+r.Host) || strings.HasPrefix(origin, "https://"+r.Host) + }) + srv.SetWSHandler(hub.Handle) + srv.SetPublishFn(hub.Broadcast) + + // Atomizer client honoring the admin base-URL override per call. + atomClient := atomize.New(func() string { + sctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if settings, err := st.GetSettings(sctx); err == nil && settings.AtomizerBaseURLOverride != "" { + return settings.AtomizerBaseURLOverride + } + return cfg.AtomizerBaseURL + }, cfg.AtomizerToken, cfg.AtomizerTimeout) + srv.SetAtomizerInfo(atomClient) + srv.AddReadinessCheck(httpx.ReadinessCheck{Name: "atomizer", Probe: atomClient.Healthy}) + + // Background job queue + atomization handlers. + runner := jobs.NewRunner(st, log, reg, 4) + atomSvc := atomize.NewService(st, atomClient, log, cfg.AppBaseURL, + []byte(cfg.SessionSecret), srv.Publish) + runner.Register(atomize.JobKindSubdivide, atomSvc.HandleSubdivide) + runner.Register(atomize.JobKindExtend, atomSvc.HandleExtend) + srv.SetJobsStatusProvider(runner) + srv.SetEnqueue(runner.Enqueue) + go runner.Run(ctx) + syncMgr := syncpkg.NewManager(st, fileStore, log, reg, func(c *domain.Customer) (syncpkg.Credentials, error) { if c.Ticketing.CredentialsEnc == "" { diff --git a/go.mod b/go.mod index 7bd3100..18bd67c 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module bountyboard go 1.26 require ( + github.com/coder/websocket v1.8.14 github.com/coreos/go-oidc/v3 v3.18.0 go.mongodb.org/mongo-driver/v2 v2.6.0 golang.org/x/crypto v0.53.0 diff --git a/go.sum b/go.sum index e1a29a8..b4be1b3 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g= +github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= github.com/coreos/go-oidc/v3 v3.18.0 h1:V9orjXynvu5wiC9SemFTWnG4F45v403aIcjWo0d41+A= github.com/coreos/go-oidc/v3 v3.18.0/go.mod h1:DYCf24+ncYi+XkIH97GY1+dqoRlbaSI26KVTCI9SrY4= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= diff --git a/internal/atomize/client.go b/internal/atomize/client.go new file mode 100644 index 0000000..9b8eae3 --- /dev/null +++ b/internal/atomize/client.go @@ -0,0 +1,153 @@ +// Package atomize is the app-side client of the Atomization Service (§5.1) +// plus the coefficient validation/normalization rules. +package atomize + +import ( + "context" + "errors" + "fmt" + "math" + "net/http" + "time" + + "bountyboard/internal/extsvc" +) + +type Attachment struct { + Name string `json:"name"` + URL string `json:"url"` + MimeType string `json:"mimeType"` +} + +type Constraints struct { + MinTasks int `json:"minTasks,omitempty"` + MaxTasks int `json:"maxTasks,omitempty"` +} + +type AtomizeRequest struct { + TaskID string `json:"taskId"` + Title string `json:"title"` + Description string `json:"description"` + AcceptanceCriteria []string `json:"acceptanceCriteria"` + Attachments []Attachment `json:"attachments"` + Links []string `json:"links"` + SubdivisionNote string `json:"subdivisionNote,omitempty"` + Constraints *Constraints `json:"constraints,omitempty"` +} + +type ExtendRequest struct { + TaskID string `json:"taskId"` + Title string `json:"title"` + Description string `json:"description"` + AcceptanceCriteria []string `json:"acceptanceCriteria"` + Attachments []Attachment `json:"attachments"` + Links []string `json:"links"` + ExtensionNote string `json:"extensionNote"` +} + +type SubTask struct { + Title string `json:"title"` + Description string `json:"description"` + AcceptanceCriteria []string `json:"acceptanceCriteria"` + EffortCoefficient float64 `json:"effortCoefficient"` +} + +type AtomizeResponse struct { + Tasks []SubTask `json:"tasks"` + Model string `json:"model,omitempty"` + Notes string `json:"notes,omitempty"` +} + +type ExtendResponse struct { + Task SubTask `json:"task"` + Model string `json:"model,omitempty"` + Notes string `json:"notes,omitempty"` +} + +// ErrBadCoefficients is the 502-class rejection when the service returns +// coefficients we are not willing to repair (§5.1). +var ErrBadCoefficients = errors.New("atomizer returned invalid effort coefficients") + +type Client struct { + c *extsvc.Client +} + +// New builds the client. baseURL is resolved per call so the admin settings +// override applies without restart. +func New(baseURL func() string, token string, timeout time.Duration) *Client { + return &Client{c: extsvc.NewClient("atomizer", baseURL, token, timeout)} +} + +func (a *Client) BreakerState() string { return a.c.BreakerState() } +func (a *Client) Healthy(ctx context.Context) error { return a.c.Healthy(ctx) } + +// Atomize calls POST /v1/atomize and applies the §5.1 contract: coefficients +// must sum to 1 ± 0.001; a deviation ≤ 0.05 is defensively renormalized, +// anything worse is rejected as a service error. +func (a *Client) Atomize(ctx context.Context, req AtomizeRequest) (*AtomizeResponse, error) { + var resp AtomizeResponse + if err := a.c.Call(ctx, http.MethodPost, "/v1/atomize", req, &resp); err != nil { + return nil, err + } + if len(resp.Tasks) == 0 { + return nil, fmt.Errorf("%w: empty task list", ErrBadCoefficients) + } + coeffs := make([]float64, len(resp.Tasks)) + for i, t := range resp.Tasks { + coeffs[i] = t.EffortCoefficient + } + normalized, err := NormalizeCoefficients(coeffs) + if err != nil { + return nil, err + } + for i := range resp.Tasks { + resp.Tasks[i].EffortCoefficient = normalized[i] + } + return &resp, nil +} + +// Extend calls POST /v1/extend. The sibling coefficient is relative to the +// source task and may exceed 1 up to 2.0 (§5.1). +func (a *Client) Extend(ctx context.Context, req ExtendRequest) (*ExtendResponse, error) { + var resp ExtendResponse + if err := a.c.Call(ctx, http.MethodPost, "/v1/extend", req, &resp); err != nil { + return nil, err + } + c := resp.Task.EffortCoefficient + if c <= 0 || c > 2.0 || math.IsNaN(c) { + return nil, fmt.Errorf("%w: extension coefficient %v outside (0, 2]", ErrBadCoefficients, c) + } + return &resp, nil +} + +// NormalizeCoefficients enforces the §5.1 sum contract. Each coefficient +// must be in (0, 1]; the sum must be 1 ± 0.001 (accepted as-is) or within +// ± 0.05 (renormalized proportionally, then rounded to 4 decimals with the +// remainder folded into the largest entry so the sum is exactly 1). +func NormalizeCoefficients(coeffs []float64) ([]float64, error) { + if len(coeffs) == 0 { + return nil, fmt.Errorf("%w: empty", ErrBadCoefficients) + } + sum := 0.0 + for _, c := range coeffs { + if math.IsNaN(c) || c <= 0 || c > 1 { + return nil, fmt.Errorf("%w: coefficient %v outside (0, 1]", ErrBadCoefficients, c) + } + sum += c + } + if math.Abs(sum-1) > 0.05 { + return nil, fmt.Errorf("%w: sum %.4f deviates more than 0.05 from 1", ErrBadCoefficients, sum) + } + out := make([]float64, len(coeffs)) + largest := 0 + total := 0.0 + for i, c := range coeffs { + out[i] = math.Round(c/sum*10000) / 10000 + total += out[i] + if out[i] > out[largest] { + largest = i + } + } + out[largest] = math.Round((out[largest]+1-total)*10000) / 10000 + return out, nil +} diff --git a/internal/atomize/client_test.go b/internal/atomize/client_test.go new file mode 100644 index 0000000..f8de41e --- /dev/null +++ b/internal/atomize/client_test.go @@ -0,0 +1,143 @@ +package atomize + +import ( + "context" + "encoding/json" + "errors" + "math" + "net/http" + "net/http/httptest" + "testing" + "time" + + "bountyboard/internal/extsvc" +) + +func TestNormalizeCoefficients(t *testing.T) { + tests := []struct { + name string + in []float64 + wantErr bool + wantSum float64 + }{ + {"exact sum kept", []float64{0.2, 0.5, 0.3}, false, 1}, + {"tiny deviation accepted", []float64{0.2, 0.5, 0.3001}, false, 1}, + {"small deviation renormalized", []float64{0.21, 0.52, 0.31}, false, 1}, // sum 1.04 + {"under-sum renormalized", []float64{0.3, 0.3, 0.36}, false, 1}, // sum 0.96 + {"large deviation rejected", []float64{0.5, 0.5, 0.5}, true, 0}, + {"zero coefficient rejected", []float64{0, 0.5, 0.5}, true, 0}, + {"negative rejected", []float64{-0.1, 0.6, 0.5}, true, 0}, + {"above one rejected", []float64{1.2}, true, 0}, + {"empty rejected", nil, true, 0}, + {"single exact", []float64{1.0}, false, 1}, + {"NaN rejected", []float64{math.NaN(), 0.5}, true, 0}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + out, err := NormalizeCoefficients(tt.in) + if tt.wantErr { + if !errors.Is(err, ErrBadCoefficients) { + t.Fatalf("want ErrBadCoefficients, got %v", err) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + sum := 0.0 + for _, c := range out { + if c <= 0 || c > 1 { + t.Errorf("normalized coefficient %v out of range", c) + } + sum += c + } + if math.Abs(sum-tt.wantSum) > 1e-9 { + t.Errorf("sum = %.6f, want exactly %v (coeffs %v)", sum, tt.wantSum, out) + } + }) + } +} + +func fakeAtomizer(t *testing.T, handler http.HandlerFunc) *Client { + t.Helper() + srv := httptest.NewServer(handler) + t.Cleanup(srv.Close) + return New(func() string { return srv.URL }, "test-token", 5*time.Second) +} + +func TestAtomizeAppliesNormalization(t *testing.T) { + client := fakeAtomizer(t, func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bearer test-token" { + t.Errorf("missing bearer token") + } + var req AtomizeRequest + json.NewDecoder(r.Body).Decode(&req) + if req.TaskID != "01JTASK" { + t.Errorf("taskId = %q", req.TaskID) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(AtomizeResponse{Tasks: []SubTask{ + {Title: "A", EffortCoefficient: 0.31}, // sum 1.03 → renormalize + {Title: "B", EffortCoefficient: 0.41}, + {Title: "C", EffortCoefficient: 0.31}, + }}) + }) + resp, err := client.Atomize(context.Background(), AtomizeRequest{TaskID: "01JTASK"}) + if err != nil { + t.Fatal(err) + } + sum := 0.0 + for _, st := range resp.Tasks { + sum += st.EffortCoefficient + } + if math.Abs(sum-1) > 1e-9 { + t.Fatalf("normalized sum = %v", sum) + } +} + +func TestAtomizeRejectsBadSum(t *testing.T) { + client := fakeAtomizer(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(AtomizeResponse{Tasks: []SubTask{ + {Title: "A", EffortCoefficient: 0.9}, + {Title: "B", EffortCoefficient: 0.9}, + }}) + }) + _, err := client.Atomize(context.Background(), AtomizeRequest{}) + if !errors.Is(err, ErrBadCoefficients) { + t.Fatalf("want ErrBadCoefficients, got %v", err) + } +} + +func TestExtendValidatesCoefficient(t *testing.T) { + coeff := 2.5 + client := fakeAtomizer(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(ExtendResponse{Task: SubTask{Title: "X", EffortCoefficient: coeff}}) + }) + if _, err := client.Extend(context.Background(), ExtendRequest{}); !errors.Is(err, ErrBadCoefficients) { + t.Fatalf("coefficient 2.5: want ErrBadCoefficients, got %v", err) + } + coeff = 1.4 // legal for extensions + if _, err := client.Extend(context.Background(), ExtendRequest{}); err != nil { + t.Fatalf("coefficient 1.4 should pass: %v", err) + } +} + +func TestBreakerOpensAfterRepeatedFailures(t *testing.T) { + client := fakeAtomizer(t, func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "boom", http.StatusInternalServerError) + }) + for i := 0; i < 3; i++ { + if _, err := client.Atomize(context.Background(), AtomizeRequest{}); err == nil { + t.Fatal("expected failure") + } + } + if client.BreakerState() != "open" { + t.Fatalf("breaker = %s, want open", client.BreakerState()) + } + _, err := client.Atomize(context.Background(), AtomizeRequest{}) + if !errors.Is(err, extsvc.ErrBreakerOpen) { + t.Fatalf("want ErrBreakerOpen, got %v", err) + } +} diff --git a/internal/atomize/service.go b/internal/atomize/service.go new file mode 100644 index 0000000..f4d2fde --- /dev/null +++ b/internal/atomize/service.go @@ -0,0 +1,263 @@ +package atomize + +import ( + "context" + "fmt" + "log/slog" + "time" + + "bountyboard/internal/domain" + "bountyboard/internal/files" + "bountyboard/internal/jobs" + "bountyboard/internal/store" +) + +// JobKindSubdivide / JobKindExtend are the queue kinds this service handles. +const ( + JobKindSubdivide = "atomize.subdivide" + JobKindExtend = "atomize.extend" +) + +type SubdividePayload struct { + TaskID string `bson:"taskId"` + Note string `bson:"note"` + MinTasks int `bson:"minTasks"` + MaxTasks int `bson:"maxTasks"` + ActorID string `bson:"actorId"` + FromStatus string `bson:"fromStatus"` // revert target on terminal failure +} + +type ExtendPayload struct { + TaskID string `bson:"taskId"` + Note string `bson:"note"` + ActorID string `bson:"actorId"` +} + +// Service executes atomization jobs: it owns the §5.1 request building, +// child/sibling task creation, status transitions, notifications, and live +// board updates. +type Service struct { + st *store.Store + client *Client + log *slog.Logger + appBaseURL string + secret []byte + publish func(channel, event string, payload any) +} + +func NewService(st *store.Store, client *Client, log *slog.Logger, + appBaseURL string, secret []byte, publish func(string, string, any)) *Service { + if publish == nil { + publish = func(string, string, any) {} + } + return &Service{st: st, client: client, log: log, + appBaseURL: appBaseURL, secret: secret, publish: publish} +} + +// requestAttachments builds §5.1 attachment refs: cached files get signed +// app URLs (fetchable without a session); uncached ones keep the upstream +// URL. +func (s *Service) requestAttachments(t *domain.Task) []Attachment { + out := make([]Attachment, 0, len(t.Attachments)) + for _, a := range t.Attachments { + url := a.URL + if a.FileID != "" { + url = files.SignedURL(s.appBaseURL, s.secret, a.FileID, time.Now().Add(files.DefaultTokenTTL)) + } + out = append(out, Attachment{Name: a.Name, URL: url, MimeType: a.MimeType}) + } + return out +} + +func taskLinks(t *domain.Task) []string { + links := append([]string{}, t.Links...) + if t.External != nil && t.External.URL != "" { + links = append(links, t.External.URL) + } + return links +} + +func (s *Service) HandleSubdivide(ctx context.Context, job *jobs.Job) error { + var p SubdividePayload + if err := jobs.DecodePayload(job, &p); err != nil { + return err + } + t, err := s.st.TaskByID(ctx, p.TaskID) + if err != nil { + return fmt.Errorf("load task: %w", err) + } + if t.Status != domain.StatusAtomizing { + s.log.Warn("subdivide job for non-atomizing task, skipping", "task", t.ID, "status", t.Status) + return nil + } + + req := AtomizeRequest{ + TaskID: t.ID, + Title: t.Title, + Description: t.Description, + AcceptanceCriteria: t.AcceptanceCriteria, + Attachments: s.requestAttachments(t), + Links: taskLinks(t), + SubdivisionNote: p.Note, + } + if p.MinTasks > 0 || p.MaxTasks > 0 { + req.Constraints = &Constraints{MinTasks: p.MinTasks, MaxTasks: p.MaxTasks} + } + + resp, err := s.client.Atomize(ctx, req) + if err != nil { + if jobs.IsFinalAttempt(job) { + s.revertSubdivide(ctx, t, &p, err) + } + return fmt.Errorf("atomize call: %w", err) + } + + // Create children, all atomized (§4.4). + childIDs := make([]string, 0, len(resp.Tasks)) + for _, sub := range resp.Tasks { + child := &domain.Task{ + CustomerID: t.CustomerID, + ConsultantID: t.ConsultantID, + Origin: domain.OriginSubdivided, + ParentID: t.ID, + RootID: t.RootID, + Title: sub.Title, + Description: sub.Description, + AcceptanceCriteria: sub.AcceptanceCriteria, + EffortCoefficient: sub.EffortCoefficient, + Budget: t.Budget, + Bounty: domain.ComputeBounty(sub.EffortCoefficient, t.Budget), + Status: domain.StatusAtomized, + Timeline: []domain.TimelineEntry{{ + At: time.Now().UTC(), ActorID: "system", Event: "created_by_subdivision", + Data: map[string]any{"parentId": t.ID, "model": resp.Model}, + }}, + } + if err := s.st.InsertTask(ctx, child); err != nil { + return fmt.Errorf("insert child: %w", err) + } + childIDs = append(childIDs, child.ID) + } + + err = s.st.TransitionTask(ctx, t, domain.StatusAtomized, p.ActorID, "subdivided", + map[string]any{"children": len(childIDs), "model": resp.Model, "notes": resp.Notes}, + map[string]any{"atomizationNote": p.Note}) + if err != nil { + // children exist; a concurrent edit raced us — reload once and retry + if fresh, ferr := s.st.TaskByID(ctx, t.ID); ferr == nil && fresh.Status == domain.StatusAtomizing { + err = s.st.TransitionTask(ctx, fresh, domain.StatusAtomized, p.ActorID, "subdivided", + map[string]any{"children": len(childIDs)}, map[string]any{"atomizationNote": p.Note}) + } + if err != nil { + return fmt.Errorf("finish subdivision: %w", err) + } + } + + s.notify(ctx, t.ConsultantID, "subdivision_ready", + fmt.Sprintf("Subdivision ready: %d subtasks", len(childIDs)), + t.Title, "/consultant/board?customerId="+t.CustomerID) + s.publish("board", "task.subdivided", map[string]any{ + "taskId": t.ID, "customerId": t.CustomerID, "childIds": childIDs, + }) + return nil +} + +func (s *Service) revertSubdivide(ctx context.Context, t *domain.Task, p *SubdividePayload, cause error) { + target := domain.TaskStatus(p.FromStatus) + if !target.Valid() { + target = domain.StatusImported + } + fresh, err := s.st.TaskByID(ctx, t.ID) + if err != nil || fresh.Status != domain.StatusAtomizing { + return + } + if err := s.st.TransitionTask(ctx, fresh, target, "system", "atomization_failed", + map[string]any{"error": cause.Error()}, nil); err != nil { + s.log.Error("revert failed subdivision", "task", t.ID, "err", err) + return + } + s.notify(ctx, t.ConsultantID, "atomization_failed", + "Subdivision failed: "+t.Title, cause.Error(), + "/consultant/board?customerId="+t.CustomerID) + s.publish("board", "task.atomization_failed", map[string]any{ + "taskId": t.ID, "customerId": t.CustomerID, + }) +} + +func (s *Service) HandleExtend(ctx context.Context, job *jobs.Job) error { + var p ExtendPayload + if err := jobs.DecodePayload(job, &p); err != nil { + return err + } + t, err := s.st.TaskByID(ctx, p.TaskID) + if err != nil { + return fmt.Errorf("load task: %w", err) + } + if t.Status != domain.StatusImported && t.Status != domain.StatusAtomized { + s.log.Warn("extend job for wrong-status task, skipping", "task", t.ID, "status", t.Status) + return nil + } + + resp, err := s.client.Extend(ctx, ExtendRequest{ + TaskID: t.ID, + Title: t.Title, + Description: t.Description, + AcceptanceCriteria: t.AcceptanceCriteria, + Attachments: s.requestAttachments(t), + Links: taskLinks(t), + ExtensionNote: p.Note, + }) + if err != nil { + if jobs.IsFinalAttempt(job) { + s.notify(ctx, t.ConsultantID, "atomization_failed", + "Extension failed: "+t.Title, err.Error(), + "/consultant/board?customerId="+t.CustomerID) + } + return fmt.Errorf("extend call: %w", err) + } + + sibling := &domain.Task{ + CustomerID: t.CustomerID, + ConsultantID: t.ConsultantID, + Origin: domain.OriginExtended, + ParentID: t.ID, // §4.4: parentId records the sibling source + RootID: t.RootID, + Title: resp.Task.Title, + Description: resp.Task.Description, + AcceptanceCriteria: resp.Task.AcceptanceCriteria, + EffortCoefficient: resp.Task.EffortCoefficient, + Budget: t.Budget, // §5.1: same budget as the source + Bounty: domain.ComputeBounty(resp.Task.EffortCoefficient, t.Budget), + Status: domain.StatusAtomized, + Timeline: []domain.TimelineEntry{{ + At: time.Now().UTC(), ActorID: "system", Event: "created_by_extension", + Data: map[string]any{"sourceId": t.ID, "model": resp.Model, "note": p.Note}, + }}, + } + if err := s.st.InsertTask(ctx, sibling); err != nil { + return fmt.Errorf("insert sibling: %w", err) + } + if err := s.st.AppendTimeline(ctx, t.ID, domain.TimelineEntry{ + ActorID: p.ActorID, Event: "extended", + Data: map[string]any{"siblingId": sibling.ID, "note": p.Note}, + }); err != nil { + s.log.Warn("append extend timeline", "err", err) + } + + s.notify(ctx, t.ConsultantID, "extension_ready", + "Extension ready: "+sibling.Title, t.Title, + "/consultant/board?customerId="+t.CustomerID) + s.publish("board", "task.extended", map[string]any{ + "taskId": t.ID, "siblingId": sibling.ID, "customerId": t.CustomerID, + }) + return nil +} + +func (s *Service) notify(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.st.InsertNotification(ctx, n); err != nil { + s.log.Warn("insert notification", "err", err) + return + } + s.publish("notifications", "notification", n) +} diff --git a/internal/atomize/service_integration_test.go b/internal/atomize/service_integration_test.go new file mode 100644 index 0000000..1e9d914 --- /dev/null +++ b/internal/atomize/service_integration_test.go @@ -0,0 +1,255 @@ +//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") + } +} diff --git a/internal/extsvc/breaker.go b/internal/extsvc/breaker.go new file mode 100644 index 0000000..30ecb46 --- /dev/null +++ b/internal/extsvc/breaker.go @@ -0,0 +1,73 @@ +// Package extsvc holds shared plumbing for the two independent external +// services (Atomization, Work Performer): a small circuit breaker and a +// retrying JSON HTTP helper. The services themselves stay fully independent +// (§5) — each gets its own breaker instance, base URL, and token. +package extsvc + +import ( + "sync" + "time" +) + +// Breaker implements §11.16: open after 3 consecutive failures, half-open +// probe every 60 s. +type Breaker struct { + mu sync.Mutex + fails int + openedAt time.Time + probing bool + now func() time.Time + threshold int + cooldown time.Duration +} + +func NewBreaker() *Breaker { + return &Breaker{now: time.Now, threshold: 3, cooldown: 60 * time.Second} +} + +// Allow reports whether a call may proceed right now. In the open state one +// probe is allowed per cooldown window (half-open). +func (b *Breaker) Allow() bool { + b.mu.Lock() + defer b.mu.Unlock() + if b.fails < b.threshold { + return true + } + if b.probing { + return false + } + if b.now().Sub(b.openedAt) >= b.cooldown { + b.probing = true // half-open: exactly one probe in flight + return true + } + return false +} + +// Record reports a call outcome. Success closes the breaker; failure counts +// toward (or re-arms) the open state. +func (b *Breaker) Record(err error) { + b.mu.Lock() + defer b.mu.Unlock() + b.probing = false + if err == nil { + b.fails = 0 + return + } + b.fails++ + if b.fails >= b.threshold { + b.openedAt = b.now() + } +} + +// State returns closed | open | half-open for status panels. +func (b *Breaker) State() string { + b.mu.Lock() + defer b.mu.Unlock() + if b.fails < b.threshold { + return "closed" + } + if b.probing || b.now().Sub(b.openedAt) >= b.cooldown { + return "half-open" + } + return "open" +} diff --git a/internal/extsvc/breaker_test.go b/internal/extsvc/breaker_test.go new file mode 100644 index 0000000..6273dd4 --- /dev/null +++ b/internal/extsvc/breaker_test.go @@ -0,0 +1,75 @@ +package extsvc + +import ( + "errors" + "testing" + "time" +) + +func TestBreakerLifecycle(t *testing.T) { + b := NewBreaker() + now := time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC) + b.now = func() time.Time { return now } + boom := errors.New("boom") + + if b.State() != "closed" || !b.Allow() { + t.Fatal("fresh breaker must be closed") + } + + // two failures: still closed + b.Record(boom) + b.Record(boom) + if b.State() != "closed" || !b.Allow() { + t.Fatal("breaker must stay closed below threshold") + } + + // third consecutive failure opens it + b.Record(boom) + if b.State() != "open" { + t.Fatalf("state = %s, want open", b.State()) + } + if b.Allow() { + t.Fatal("open breaker must reject calls") + } + + // after the cooldown one probe is allowed (half-open) + now = now.Add(61 * time.Second) + if !b.Allow() { + t.Fatal("half-open must allow one probe") + } + if b.Allow() { + t.Fatal("only one probe may be in flight") + } + if b.State() != "half-open" { + t.Fatalf("state = %s, want half-open", b.State()) + } + + // failed probe re-opens for another cooldown + b.Record(boom) + if b.Allow() { + t.Fatal("failed probe must re-open the breaker") + } + now = now.Add(61 * time.Second) + if !b.Allow() { + t.Fatal("next probe window") + } + + // successful probe closes it fully + b.Record(nil) + if b.State() != "closed" || !b.Allow() || !b.Allow() { + t.Fatal("successful probe must close the breaker") + } +} + +func TestBreakerSuccessResetsCount(t *testing.T) { + b := NewBreaker() + boom := errors.New("x") + b.Record(boom) + b.Record(boom) + b.Record(nil) // non-consecutive failures never open + b.Record(boom) + b.Record(boom) + if b.State() != "closed" { + t.Fatal("interleaved success must reset the failure count") + } +} diff --git a/internal/extsvc/client.go b/internal/extsvc/client.go new file mode 100644 index 0000000..cdb0c1c --- /dev/null +++ b/internal/extsvc/client.go @@ -0,0 +1,150 @@ +package extsvc + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "math/rand/v2" + "net/http" + "time" +) + +// ErrBreakerOpen is returned without touching the network when the circuit +// is open. +var ErrBreakerOpen = errors.New("circuit breaker open") + +// ServiceError is a non-2xx response from the external service, carrying +// its §5 error envelope when present. +type ServiceError struct { + Status int + Code string + Message string +} + +func (e *ServiceError) Error() string { + return fmt.Sprintf("service responded %d: %s %s", e.Status, e.Code, e.Message) +} + +// Client is a bearer-token JSON client with retry (×2 on 5xx/network with +// jitter, §12) and a circuit breaker, shared by the two service clients. +type Client struct { + Name string + BaseURL func() string // resolved per call (admin override support) + Token string + HTTP *http.Client + Breaker *Breaker +} + +func NewClient(name string, baseURL func() string, token string, timeout time.Duration) *Client { + return &Client{ + Name: name, + BaseURL: baseURL, + Token: token, + HTTP: &http.Client{Timeout: timeout}, + Breaker: NewBreaker(), + } +} + +func (c *Client) BreakerState() string { return c.Breaker.State() } + +// Healthy probes /healthz without breaker accounting (used by readyz and the +// admin panel). +func (c *Client) Healthy(ctx context.Context) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.BaseURL()+"/healthz", nil) + if err != nil { + return err + } + resp, err := c.HTTP.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("%s healthz: status %d", c.Name, resp.StatusCode) + } + return nil +} + +// Call POSTs/GETs JSON through the breaker with retries. A nil out skips +// decoding. +func (c *Client) Call(ctx context.Context, method, path string, in, out any) error { + if !c.Breaker.Allow() { + return fmt.Errorf("%s: %w", c.Name, ErrBreakerOpen) + } + err := c.callOnce(ctx, method, path, in, out) + c.Breaker.Record(err) + return err +} + +func (c *Client) callOnce(ctx context.Context, method, path string, in, out any) error { + var body []byte + if in != nil { + var err error + if body, err = json.Marshal(in); err != nil { + return fmt.Errorf("encode request: %w", err) + } + } + + var lastErr error + for attempt := 0; attempt < 3; attempt++ { + if attempt > 0 { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(time.Duration(attempt)*time.Second + time.Duration(rand.IntN(500))*time.Millisecond): + } + } + req, err := http.NewRequestWithContext(ctx, method, c.BaseURL()+path, bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+c.Token) + resp, err := c.HTTP.Do(req) + if err != nil { + lastErr = fmt.Errorf("%s %s: %w", c.Name, path, err) + continue // network error → retry + } + data, err := io.ReadAll(io.LimitReader(resp.Body, 10<<20)) + resp.Body.Close() + if err != nil { + lastErr = fmt.Errorf("%s %s: read body: %w", c.Name, path, err) + continue + } + if resp.StatusCode >= 500 { + lastErr = &ServiceError{Status: resp.StatusCode, Message: string(truncate(data, 200))} + continue // 5xx → retry + } + if resp.StatusCode >= 400 { + se := &ServiceError{Status: resp.StatusCode} + var env struct { + Error struct { + Code string `json:"code"` + Message string `json:"message"` + } `json:"error"` + } + if json.Unmarshal(data, &env) == nil { + se.Code, se.Message = env.Error.Code, env.Error.Message + } + return se // 4xx is not retryable + } + if out == nil { + return nil + } + if err := json.Unmarshal(data, out); err != nil { + return fmt.Errorf("%s %s: decode response: %w", c.Name, path, err) + } + return nil + } + return lastErr +} + +func truncate(b []byte, n int) []byte { + if len(b) <= n { + return b + } + return b[:n] +} diff --git a/internal/http/providers.go b/internal/http/providers.go index 4c3b5a7..9e8cea1 100644 --- a/internal/http/providers.go +++ b/internal/http/providers.go @@ -1,6 +1,10 @@ package httpx -import "context" +import ( + "context" + "errors" + "net/http" +) // Late-bound providers: these are wired by later subsystems (atomizer // client, work-performer client, sync manager, job runner) after the server @@ -35,6 +39,37 @@ func (s *Server) Publish(channel, event string, payload any) { // SetPublishFn wires the actual hub broadcast. func (s *Server) SetPublishFn(f func(channel, event string, payload any)) { s.publishFn = f } + +// SetEnqueue wires the background job queue. +func (s *Server) SetEnqueue(f func(ctx context.Context, kind string, payload any) (string, error)) { + s.enqueueFn = f +} + +func (s *Server) enqueue(ctx context.Context, kind string, payload any) (string, error) { + if s.enqueueFn == nil { + return "", errors.New("job queue is not running") + } + return s.enqueueFn(ctx, kind, payload) +} + +// SetWSHandler wires the WebSocket hub's connection handler. +func (s *Server) SetWSHandler(f func(w http.ResponseWriter, r *http.Request, userID string)) { + s.wsHandler = f +} + +// handleWS authenticates the session and hands the connection to the hub. +func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) { + u := s.pageUser(r) + if u == nil { + writeError(w, http.StatusUnauthorized, "unauthenticated", "session required") + return + } + if s.wsHandler == nil { + writeError(w, http.StatusServiceUnavailable, "ws_unavailable", "live updates are not running") + return + } + s.wsHandler(w, r, u.ID) +} func (s *Server) SetSyncStatusProvider(p StatusProvider) { s.syncProvider = p } diff --git a/internal/http/server.go b/internal/http/server.go index 0cef178..f914f00 100644 --- a/internal/http/server.go +++ b/internal/http/server.go @@ -39,6 +39,8 @@ type Server struct { 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) } func New(cfg *config.Config, log *slog.Logger, reg *metrics.Registry, st *store.Store, fs *files.Store) *Server { @@ -67,6 +69,8 @@ func New(cfg *config.Config, log *slog.Logger, reg *metrics.Registry, st *store. s.routesAuth(mux) s.routesProfile(mux) s.routesAdmin(mux) + s.routesTasks(mux) + mux.HandleFunc("GET /ws", s.handleWS) s.routesWeb(mux) s.httpSrv = &http.Server{ diff --git a/internal/http/tasks.go b/internal/http/tasks.go new file mode 100644 index 0000000..11cd37a --- /dev/null +++ b/internal/http/tasks.go @@ -0,0 +1,467 @@ +package httpx + +import ( + "context" + "errors" + "net/http" + "strings" + "time" + + "go.mongodb.org/mongo-driver/v2/bson" + + "bountyboard/internal/atomize" + "bountyboard/internal/domain" + "bountyboard/internal/store" +) + +func (s *Server) routesTasks(mux *http.ServeMux) { + mux.Handle("GET /api/v1/consultant/board", s.authedRole("consultant", s.handleConsultantBoard)) + mux.Handle("POST /api/v1/tasks/{id}/subdivide", s.authedRole("consultant", s.handleSubdivide)) + mux.Handle("POST /api/v1/tasks/{id}/extend", s.authedRole("consultant", s.handleExtend)) + mux.Handle("PATCH /api/v1/tasks/{id}", s.authedRole("consultant", s.handleEditTask)) + mux.Handle("POST /api/v1/tasks/{id}/publish", s.authedRole("consultant", s.handlePublishOne)) + mux.Handle("POST /api/v1/tasks/publish", s.authedRole("consultant", s.handlePublishBulk)) + mux.Handle("POST /api/v1/tasks/{id}/archive", s.authed(s.handleArchiveTask)) + mux.Handle("GET /api/v1/tasks/{id}", s.requireAuth(http.HandlerFunc(s.handleTaskDetail))) + mux.Handle("GET /api/v1/service-health", s.requireAuth(http.HandlerFunc(s.handleServiceHealth))) +} + +// consultantTask loads a task and authorizes the current user: admins and +// every consultant assigned to the task's customer may manage it (§4.4). +func (s *Server) consultantTask(w http.ResponseWriter, r *http.Request, id string) (*domain.Task, *domain.Customer, 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, nil, false + } + c, err := s.store.CustomerByID(r.Context(), t.CustomerID) + if err != nil { + s.internalError(w, r, "load customer", err) + return nil, nil, false + } + u := CurrentUser(r.Context()) + if !u.Roles.Admin && !(u.Roles.Consultant && c.HasConsultant(u.ID)) { + writeError(w, http.StatusForbidden, "forbidden", "you are not assigned to this customer") + return nil, nil, false + } + return t, c, true +} + +func (s *Server) handleConsultantBoard(w http.ResponseWriter, r *http.Request) { + u := CurrentUser(r.Context()) + consultantID := u.ID + if u.Roles.Admin { + consultantID = "" // admins see all customers + } + customers, err := s.store.ListCustomers(r.Context(), consultantID, false) + if err != nil { + s.internalError(w, r, "list customers", err) + return + } + customerID := r.URL.Query().Get("customerId") + ids := []string{} + for _, c := range customers { + if customerID == "" || c.ID == customerID { + ids = append(ids, c.ID) + } + } + filter := store.TaskFilter{ + CustomerIDs: ids, + Statuses: []domain.TaskStatus{ + domain.StatusImported, domain.StatusAtomizing, domain.StatusAtomized, + domain.StatusPublished, domain.StatusClaimRequested, + }, + Limit: 500, + Sort: "updated", + } + if st := r.URL.Query().Get("status"); st != "" { + filter.Statuses = []domain.TaskStatus{domain.TaskStatus(st)} + } + tasks := []domain.Task{} + if len(ids) > 0 { + if tasks, err = s.store.ListTasks(r.Context(), filter); err != nil { + s.internalError(w, r, "list tasks", err) + return + } + } + out := make([]map[string]any, len(customers)) + for i := range customers { + out[i] = customerJSON(&customers[i]) + } + writeJSON(w, http.StatusOK, map[string]any{"customers": out, "tasks": tasks}) +} + +func (s *Server) handleSubdivide(w http.ResponseWriter, r *http.Request) { + var req struct { + Note string `json:"note"` + Constraints *struct { + MinTasks int `json:"minTasks"` + MaxTasks int `json:"maxTasks"` + } `json:"constraints"` + ConfirmReplace bool `json:"confirmReplace"` + } + if !decodeJSON(w, r, &req) { + return + } + t, _, ok := s.consultantTask(w, r, r.PathValue("id")) + if !ok { + return + } + if t.Status != domain.StatusImported && t.Status != domain.StatusAtomized { + writeError(w, http.StatusConflict, "bad_status", + "only imported or atomized tasks can be subdivided (current: "+string(t.Status)+")") + return + } + minT, maxT := 0, 0 + if req.Constraints != nil { + minT, maxT = req.Constraints.MinTasks, req.Constraints.MaxTasks + if minT < 0 || maxT < 0 || (maxT > 0 && minT > maxT) || maxT > 20 { + writeError(w, http.StatusBadRequest, "bad_constraints", "constraints must satisfy 0 <= min <= max <= 20") + return + } + } + + // Re-run subdivision replaces unpublished children after confirmation + // (§11.7). + children, err := s.store.ListTasks(r.Context(), store.TaskFilter{ParentID: t.ID, Limit: 500}) + if err != nil { + s.internalError(w, r, "list children", err) + return + } + var unpublished []string + for _, c := range children { + if c.Status == domain.StatusAtomized && c.Origin == domain.OriginSubdivided { + unpublished = append(unpublished, c.ID) + } + } + if len(children) > 0 && !req.ConfirmReplace { + writeError(w, http.StatusConflict, "confirm_replace", + "this task already has subtasks; re-running replaces the unpublished ones — confirm to proceed") + return + } + if len(unpublished) > 0 { + if _, err := s.store.DeleteTasks(r.Context(), unpublished); err != nil { + s.internalError(w, r, "delete unpublished children", err) + return + } + } + + fromStatus := string(t.Status) + if err := s.store.TransitionTask(r.Context(), t, domain.StatusAtomizing, + CurrentUser(r.Context()).ID, "subdivide_requested", + map[string]any{"note": req.Note}, nil); err != nil { + s.taskTransitionError(w, err) + return + } + jobID, err := s.enqueue(r.Context(), atomize.JobKindSubdivide, atomize.SubdividePayload{ + TaskID: t.ID, Note: req.Note, MinTasks: minT, MaxTasks: maxT, + ActorID: CurrentUser(r.Context()).ID, FromStatus: fromStatus, + }) + if err != nil { + s.internalError(w, r, "enqueue subdivide", err) + return + } + s.Publish("board", "task.atomizing", map[string]any{"taskId": t.ID, "customerId": t.CustomerID}) + writeJSON(w, http.StatusAccepted, map[string]string{"jobId": jobID, "status": "queued"}) +} + +func (s *Server) handleExtend(w http.ResponseWriter, r *http.Request) { + var req struct { + Note string `json:"note"` + } + if !decodeJSON(w, r, &req) { + return + } + if strings.TrimSpace(req.Note) == "" { + writeError(w, http.StatusBadRequest, "note_required", "extension requires a note (§1.1)") + return + } + t, _, ok := s.consultantTask(w, r, r.PathValue("id")) + if !ok { + return + } + if t.Status != domain.StatusImported && t.Status != domain.StatusAtomized { + writeError(w, http.StatusConflict, "bad_status", + "only imported or atomized tasks can be extended (current: "+string(t.Status)+")") + return + } + jobID, err := s.enqueue(r.Context(), atomize.JobKindExtend, atomize.ExtendPayload{ + TaskID: t.ID, Note: req.Note, ActorID: CurrentUser(r.Context()).ID, + }) + if err != nil { + s.internalError(w, r, "enqueue extend", err) + return + } + writeJSON(w, http.StatusAccepted, map[string]string{"jobId": jobID, "status": "queued"}) +} + +func (s *Server) handleEditTask(w http.ResponseWriter, r *http.Request) { + var req struct { + Version *int64 `json:"version"` + Title *string `json:"title"` + Description *string `json:"description"` + AcceptanceCriteria *[]string `json:"acceptanceCriteria"` + Links *[]string `json:"links"` + EffortCoefficient *float64 `json:"effortCoefficient"` + Budget *float64 `json:"budget"` + CascadeBudget bool `json:"cascadeBudget"` + } + if !decodeJSON(w, r, &req) { + return + } + t, _, ok := s.consultantTask(w, r, r.PathValue("id")) + if !ok { + return + } + // §6.1: bounty is frozen once approved; archived tasks are immutable. + if t.Status == domain.StatusApproved || t.Status == domain.StatusArchived { + writeError(w, http.StatusConflict, "immutable", "approved/archived tasks cannot be edited") + return + } + + set := bson.M{} + coeff, budget := t.EffortCoefficient, t.Budget + if req.Title != nil && strings.TrimSpace(*req.Title) != "" { + set["title"] = strings.TrimSpace(*req.Title) + } + if req.Description != nil { + set["description"] = *req.Description + } + if req.AcceptanceCriteria != nil { + set["acceptanceCriteria"] = *req.AcceptanceCriteria + } + if req.Links != nil { + set["links"] = *req.Links + } + if req.EffortCoefficient != nil { + coeff = *req.EffortCoefficient + maxCoeff := 1.0 + if t.Origin == domain.OriginExtended { + maxCoeff = 2.0 // §5.1: extensions may exceed 1 + } + if coeff <= 0 || coeff > maxCoeff { + writeError(w, http.StatusBadRequest, "bad_coefficient", + "effort coefficient out of range") + return + } + set["effortCoefficient"] = coeff + } + if req.Budget != nil { + budget = *req.Budget + if budget < 0 { + writeError(w, http.StatusBadRequest, "bad_budget", "budget must be >= 0") + return + } + set["budget"] = budget + } + if len(set) == 0 { + writeError(w, http.StatusBadRequest, "empty_update", "no recognized fields") + return + } + set["bounty"] = domain.ComputeBounty(coeff, budget) + + version := t.Version + if req.Version != nil { + version = *req.Version + } + err := store.UpdateVersioned(r.Context(), s.store.DB.Collection("tasks"), t.ID, version, bson.M{ + "$set": set, + "$push": bson.M{"timeline": domain.TimelineEntry{ + At: time.Now().UTC(), ActorID: CurrentUser(r.Context()).ID, Event: "edited", + }}, + }) + if err != nil { + if errors.Is(err, store.ErrVersionConflict) { + writeError(w, http.StatusConflict, "conflict", "task was modified concurrently — reload") + return + } + s.internalError(w, r, "edit task", err) + return + } + + // Optional budget cascade to descendants whose bounty is not yet frozen. + if req.Budget != nil && req.CascadeBudget { + if err := s.cascadeBudget(r.Context(), t.ID, budget); err != nil { + s.log.Warn("cascade budget", "task", t.ID, "err", err) + } + } + + fresh, err := s.store.TaskByID(r.Context(), t.ID) + if err != nil { + s.internalError(w, r, "reload task", err) + return + } + s.Publish("board", "task.updated", map[string]any{"taskId": t.ID, "customerId": t.CustomerID}) + writeJSON(w, http.StatusOK, map[string]any{"task": fresh}) +} + +// cascadeBudget walks descendants breadth-first updating budget + bounty for +// tasks that are not approved/archived. +func (s *Server) cascadeBudget(ctx context.Context, rootID string, budget float64) error { + frontier := []string{rootID} + for len(frontier) > 0 { + children, err := s.store.ListTasks(ctx, store.TaskFilter{ParentID: frontier[0], Limit: 500}) + if err != nil { + return err + } + frontier = frontier[1:] + for _, c := range children { + if c.Status == domain.StatusApproved || c.Status == domain.StatusArchived { + continue + } + _, err := s.store.DB.Collection("tasks").UpdateOne(ctx, + bson.M{"_id": c.ID}, + bson.M{"$set": bson.M{ + "budget": budget, + "bounty": domain.ComputeBounty(c.EffortCoefficient, budget), + "updatedAt": time.Now().UTC(), + }, "$inc": bson.M{"version": 1}}) + if err != nil { + return err + } + frontier = append(frontier, c.ID) + } + } + return nil +} + +func (s *Server) publishTask(ctx context.Context, t *domain.Task, actorID string) error { + return s.store.TransitionTask(ctx, t, domain.StatusPublished, actorID, "published", nil, + map[string]any{"publishedAt": time.Now().UTC()}) +} + +func (s *Server) handlePublishOne(w http.ResponseWriter, r *http.Request) { + t, _, ok := s.consultantTask(w, r, r.PathValue("id")) + if !ok { + return + } + if err := s.publishTask(r.Context(), t, CurrentUser(r.Context()).ID); err != nil { + s.taskTransitionError(w, err) + return + } + s.Publish("board", "task.published", map[string]any{"taskId": t.ID, "customerId": t.CustomerID}) + fresh, _ := s.store.TaskByID(r.Context(), t.ID) + writeJSON(w, http.StatusOK, map[string]any{"task": fresh}) +} + +func (s *Server) handlePublishBulk(w http.ResponseWriter, r *http.Request) { + var req struct { + IDs []string `json:"ids"` + } + if !decodeJSON(w, r, &req) { + return + } + if len(req.IDs) == 0 || len(req.IDs) > 100 { + writeError(w, http.StatusBadRequest, "bad_request", "ids must contain 1..100 entries") + return + } + published := []string{} + failed := map[string]string{} + for _, id := range req.IDs { + t, err := s.store.TaskByID(r.Context(), id) + if err != nil { + failed[id] = "not found" + continue + } + c, err := s.store.CustomerByID(r.Context(), t.CustomerID) + if err != nil { + failed[id] = "customer missing" + continue + } + u := CurrentUser(r.Context()) + if !u.Roles.Admin && !c.HasConsultant(u.ID) { + failed[id] = "not your customer" + continue + } + if err := s.publishTask(r.Context(), t, u.ID); err != nil { + failed[id] = err.Error() + continue + } + published = append(published, id) + s.Publish("board", "task.published", map[string]any{"taskId": id, "customerId": t.CustomerID}) + } + writeJSON(w, http.StatusOK, map[string]any{"published": published, "failed": failed}) +} + +func (s *Server) handleArchiveTask(w http.ResponseWriter, r *http.Request) { + u := CurrentUser(r.Context()) + if !u.Roles.Admin && !u.Roles.Consultant { + writeError(w, http.StatusForbidden, "forbidden", "requires consultant or admin role") + return + } + t, _, ok := s.consultantTask(w, r, r.PathValue("id")) + if !ok { + return + } + if err := s.store.TransitionTask(r.Context(), t, domain.StatusArchived, u.ID, "archived", nil, nil); err != nil { + s.taskTransitionError(w, err) + return + } + s.Publish("board", "task.archived", map[string]any{"taskId": t.ID, "customerId": t.CustomerID}) + w.WriteHeader(http.StatusNoContent) +} + +// handleTaskDetail is role-scoped: admins and customer consultants always; +// developers when they are the assignee, have a claim, or the task is on the +// public board (published/claim_requested). +func (s *Server) handleTaskDetail(w http.ResponseWriter, r *http.Request) { + t, err := s.store.TaskByID(r.Context(), r.PathValue("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 + } + u := CurrentUser(r.Context()) + allowed := u.Roles.Admin + if !allowed && u.Roles.Consultant { + if c, err := s.store.CustomerByID(r.Context(), t.CustomerID); err == nil && c.HasConsultant(u.ID) { + allowed = true + } + } + if !allowed && u.Roles.Developer { + allowed = t.IsAssignedTo(u.ID) || t.HasClaimFrom(u.ID) || + t.Status == domain.StatusPublished || t.Status == domain.StatusClaimRequested + } + if !allowed { + writeError(w, http.StatusForbidden, "forbidden", "no access to this task") + return + } + writeJSON(w, http.StatusOK, map[string]any{"task": t}) +} + +// handleServiceHealth powers UI gating (§11.16): disable Subdivide/Extend +// when the atomizer is down, independently of the work performer. +func (s *Server) handleServiceHealth(w http.ResponseWriter, r *http.Request) { + ctx, cancel := context.WithTimeout(r.Context(), 4*time.Second) + defer cancel() + atom := s.probeExternal(ctx, "atomizer", s.effectiveAtomizerBaseURL(ctx)) + perf := s.probeExternal(ctx, "work-performer", s.cfg.WorkPerformerBaseURL) + if s.atomizer != nil { + atom.Breaker = s.atomizer.BreakerState() + } + if s.performer != nil { + perf.Breaker = s.performer.BreakerState() + } + writeJSON(w, http.StatusOK, map[string]any{"atomizer": atom, "workPerformer": perf}) +} + +func (s *Server) taskTransitionError(w http.ResponseWriter, err error) { + var bad *domain.ErrBadTransition + switch { + case errors.As(err, &bad): + writeError(w, http.StatusConflict, "bad_status", bad.Error()) + case errors.Is(err, store.ErrVersionConflict): + writeError(w, http.StatusConflict, "conflict", "task was modified concurrently — reload") + case errors.Is(err, store.ErrNotFound): + writeError(w, http.StatusNotFound, "not_found", "task not found") + default: + writeError(w, http.StatusInternalServerError, "internal", "internal server error") + } +} diff --git a/internal/http/web.go b/internal/http/web.go index bd5bc3f..fe4ba8a 100644 --- a/internal/http/web.go +++ b/internal/http/web.go @@ -181,11 +181,21 @@ func (s *Server) routesWeb(mux *http.ServeMux) { }) })) + mux.HandleFunc("GET /consultant/board", s.page(func(w http.ResponseWriter, r *http.Request, u *domain.User) { + if !u.Roles.Consultant && !u.Roles.Admin { + s.render(w, r, "placeholder.html", &pageData{Title: "Not authorized", User: u}) + return + } + s.render(w, r, "atomization.html", &pageData{ + Title: "Atomization Board", Active: "atomization", User: u, + Scripts: []string{"/static/js/atomization.js"}, + }) + })) + // Placeholders for areas built in later phases; replaced as they land. placeholders := map[string]string{ "/board": "Bounty Board", "/my-tasks": "My Tasks", - "/consultant/board": "Atomization Board", "/consultant/reviews": "Review Queue", "/consultant/pool": "Developer Pool", "/messages": "Messages", diff --git a/internal/jobs/jobs.go b/internal/jobs/jobs.go new file mode 100644 index 0000000..20d8b2e --- /dev/null +++ b/internal/jobs/jobs.go @@ -0,0 +1,267 @@ +// Package jobs is a small persisted background job queue (§12): jobs live in +// the `jobs` collection so restarts don't lose work; workers recover from +// panics and retry with exponential backoff up to 3 attempts. +package jobs + +import ( + "context" + "errors" + "fmt" + "log/slog" + "runtime/debug" + "time" + + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" + + "bountyboard/internal/metrics" + "bountyboard/internal/store" + "bountyboard/internal/ulid" +) + +const ( + StatusQueued = "queued" + StatusRunning = "running" + StatusSucceeded = "succeeded" + StatusFailed = "failed" + + maxAttempts = 3 + // a "running" job untouched this long is presumed lost to a crash and + // re-queued + staleAfter = 5 * time.Minute +) + +type Job struct { + ID string `bson:"_id" json:"id"` + Kind string `bson:"kind" json:"kind"` + Payload bson.Raw `bson:"payload" json:"-"` + Status string `bson:"status" json:"status"` + Attempts int `bson:"attempts" json:"attempts"` + RunAfter time.Time `bson:"runAfter" json:"runAfter"` + LastError string `bson:"lastError,omitempty" json:"lastError,omitempty"` + CreatedAt time.Time `bson:"createdAt" json:"createdAt"` + UpdatedAt time.Time `bson:"updatedAt" json:"updatedAt"` +} + +// Handler processes one job. err == nil marks the job succeeded; an error +// schedules a retry (with backoff) until attempts run out — the final +// failure is reported via OnExhausted if the handler set one. +type Handler func(ctx context.Context, job *Job) error + +type Runner struct { + st *store.Store + log *slog.Logger + reg *metrics.Registry + handlers map[string]Handler + workers int +} + +func NewRunner(st *store.Store, log *slog.Logger, reg *metrics.Registry, workers int) *Runner { + if workers < 1 { + workers = 2 + } + return &Runner{ + st: st, log: log, reg: reg, + handlers: map[string]Handler{}, workers: workers, + } +} + +func (r *Runner) coll() *mongo.Collection { return r.st.DB.Collection("jobs") } + +// Register installs the handler for a job kind (call before Run). +func (r *Runner) Register(kind string, h Handler) { r.handlers[kind] = h } + +// Enqueue persists a job for asynchronous execution. +func (r *Runner) Enqueue(ctx context.Context, kind string, payload any) (string, error) { + raw, err := bson.Marshal(payload) + if err != nil { + return "", fmt.Errorf("encode job payload: %w", err) + } + now := time.Now().UTC() + job := Job{ + ID: ulid.New(), Kind: kind, Payload: raw, + Status: StatusQueued, RunAfter: now, CreatedAt: now, UpdatedAt: now, + } + if _, err := r.coll().InsertOne(ctx, job); err != nil { + return "", fmt.Errorf("enqueue %s: %w", kind, err) + } + r.reg.Inc("jobs_enqueued_total", 1) + return job.ID, nil +} + +// DecodePayload unmarshals the job payload into dst. +func DecodePayload(job *Job, dst any) error { + if err := bson.Unmarshal(job.Payload, dst); err != nil { + return fmt.Errorf("decode payload of job %s (%s): %w", job.ID, job.Kind, err) + } + return nil +} + +// Run starts the worker pool and blocks until ctx is canceled. +func (r *Runner) Run(ctx context.Context) { + for i := 0; i < r.workers; i++ { + go r.workLoop(ctx) + } + // stale-job janitor + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + r.requeueStale(ctx) + r.updateGauges(ctx) + } + } +} + +func (r *Runner) workLoop(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + default: + } + job, err := r.claim(ctx) + if err != nil { + if !errors.Is(err, context.Canceled) { + r.log.Error("claim job", "err", err) + } + sleep(ctx, 2*time.Second) + continue + } + if job == nil { + sleep(ctx, time.Second) + continue + } + r.execute(ctx, job) + } +} + +func sleep(ctx context.Context, d time.Duration) { + select { + case <-ctx.Done(): + case <-time.After(d): + } +} + +// claim atomically moves one due job to running. +func (r *Runner) claim(ctx context.Context) (*Job, error) { + now := time.Now().UTC() + var job Job + err := r.coll().FindOneAndUpdate(ctx, + bson.M{"status": StatusQueued, "runAfter": bson.M{"$lte": now}}, + bson.M{"$set": bson.M{"status": StatusRunning, "updatedAt": now}, + "$inc": bson.M{"attempts": 1}}, + ).Decode(&job) + if errors.Is(err, mongo.ErrNoDocuments) { + return nil, nil + } + if err != nil { + return nil, err + } + job.Attempts++ // FindOneAndUpdate returned the pre-update doc + job.Status = StatusRunning + return &job, nil +} + +func (r *Runner) execute(ctx context.Context, job *Job) { + handler, ok := r.handlers[job.Kind] + if !ok { + r.finish(ctx, job, fmt.Errorf("no handler registered for kind %q", job.Kind), false) + return + } + var err error + func() { + defer func() { + if rec := recover(); rec != nil { + err = fmt.Errorf("panic: %v", rec) + r.log.Error("job panic", "job", job.ID, "kind", job.Kind, + "panic", fmt.Sprint(rec), "stack", string(debug.Stack())) + } + }() + jctx, cancel := context.WithTimeout(ctx, 10*time.Minute) + defer cancel() + err = handler(jctx, job) + }() + r.finish(ctx, job, err, true) +} + +func (r *Runner) finish(ctx context.Context, job *Job, err error, retryable bool) { + // Bookkeeping must survive shutdown/cancellation: otherwise a canceled + // run leaves the job stuck in "running" until stale recovery. + ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + defer cancel() + now := time.Now().UTC() + switch { + case err == nil: + _, uerr := r.coll().UpdateOne(ctx, bson.M{"_id": job.ID}, bson.M{ + "$set": bson.M{"status": StatusSucceeded, "lastError": "", "updatedAt": now}}) + if uerr != nil { + r.log.Error("mark job succeeded", "job", job.ID, "err", uerr) + } + r.reg.Inc("jobs_succeeded_total", 1) + case retryable && job.Attempts < maxAttempts: + backoff := time.Duration(1<= maxAttempts } + +func (r *Runner) requeueStale(ctx context.Context) { + cutoff := time.Now().UTC().Add(-staleAfter) + res, err := r.coll().UpdateMany(ctx, + bson.M{"status": StatusRunning, "updatedAt": bson.M{"$lt": cutoff}}, + bson.M{"$set": bson.M{"status": StatusQueued, "updatedAt": time.Now().UTC()}}) + if err != nil { + r.log.Error("requeue stale jobs", "err", err) + return + } + if res.ModifiedCount > 0 { + r.log.Warn("requeued stale running jobs", "count", res.ModifiedCount) + } +} + +func (r *Runner) updateGauges(ctx context.Context) { + for _, status := range []string{StatusQueued, StatusRunning, StatusFailed} { + n, err := r.coll().CountDocuments(ctx, bson.M{"status": status}) + if err == nil { + r.reg.Set("jobs_"+status, n) + } + } +} + +// Statuses feeds the admin service-status panel. +func (r *Runner) Statuses() any { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + out := map[string]int64{} + for _, status := range []string{StatusQueued, StatusRunning, StatusSucceeded, StatusFailed} { + n, err := r.coll().CountDocuments(ctx, bson.M{"status": status}) + if err != nil { + return map[string]string{"error": err.Error()} + } + out[status] = n + } + return out +} diff --git a/internal/store/indexes.go b/internal/store/indexes.go index 16a5541..0eb0470 100644 --- a/internal/store/indexes.go +++ b/internal/store/indexes.go @@ -33,8 +33,13 @@ func EnsureIndexes(ctx context.Context, db *mongo.Database) error { {Keys: bson.D{{Key: "assignee.userId", Value: 1}, {Key: "status", Value: 1}}}, {Keys: bson.D{{Key: "rootId", Value: 1}}}, {Keys: bson.D{{Key: "parentId", Value: 1}}}, + // The spec calls this "unique sparse", but a compound sparse index + // matches any doc where customerId exists (i.e. every task), which + // breaks subdivided children. A partial index expresses the intent: + // uniqueness only for tasks that actually have an external ref. {Keys: bson.D{{Key: "external.system", Value: 1}, {Key: "external.key", Value: 1}, {Key: "customerId", Value: 1}}, - Options: options.Index().SetUnique(true).SetSparse(true)}, + Options: options.Index().SetUnique(true).SetName("uniq_external_ref"). + SetPartialFilterExpression(bson.M{"external.key": bson.M{"$exists": true}})}, {Keys: bson.D{{Key: "title", Value: "text"}, {Key: "description", Value: "text"}}}, }, "bountyAwards": { @@ -63,6 +68,10 @@ func EnsureIndexes(ctx context.Context, db *mongo.Database) error { }, } + // migration: an earlier build created the external-ref index as sparse + // under the auto-generated name; drop it so the partial variant applies + _ = db.Collection("tasks").Indexes().DropOne(ctx, "external.system_1_external.key_1_customerId_1") + for coll, models := range plan { if _, err := db.Collection(coll).Indexes().CreateMany(ctx, models); err != nil { return fmt.Errorf("create indexes for %s: %w", coll, err) diff --git a/internal/store/store_integration_test.go b/internal/store/store_integration_test.go index 1f3804f..efc5c8e 100644 --- a/internal/store/store_integration_test.go +++ b/internal/store/store_integration_test.go @@ -28,7 +28,7 @@ func TestEnsureIndexesIsIdempotent(t *testing.T) { // Spot-check a few critical indexes actually exist. checks := map[string]string{ "users": "email_1", - "tasks": "external.system_1_external.key_1_customerId_1", + "tasks": "uniq_external_ref", "sessions": "expiresAt_1", } for coll, wantName := range checks { @@ -44,10 +44,10 @@ func TestEnsureIndexesIsIdempotent(t *testing.T) { for _, s := range specs { if s["name"] == wantName { found = true - // the unique upsert key must really be unique+sparse + // the upsert key must be unique and partial (external-only) if coll == "tasks" { - if s["unique"] != true || s["sparse"] != true { - t.Errorf("%s/%s: want unique sparse, got %v", coll, wantName, s) + if s["unique"] != true || s["partialFilterExpression"] == nil { + t.Errorf("%s/%s: want unique partial, got %v", coll, wantName, s) } } } diff --git a/internal/ws/hub.go b/internal/ws/hub.go new file mode 100644 index 0000000..57903eb --- /dev/null +++ b/internal/ws/hub.go @@ -0,0 +1,190 @@ +// Package ws implements the single multiplexed live-update WebSocket (§2.2): +// one connection per logged-in session carrying `chat`, `board`, and +// `notifications` channels, with 30 s heartbeats so idle proxies keep the +// connection alive (§12). +package ws + +import ( + "context" + "encoding/json" + "log/slog" + "net/http" + "sync" + "time" + + "github.com/coder/websocket" +) + +// Message is the wire format in both directions. +type Message struct { + Channel string `json:"channel"` + Event string `json:"event"` + Data json.RawMessage `json:"data,omitempty"` +} + +type client struct { + userID string + send chan []byte +} + +// Inbound handles client→server messages (e.g. typing indicators). +type Inbound func(userID string, msg Message) + +type Hub struct { + log *slog.Logger + originOK func(r *http.Request) bool + inbound Inbound + mu sync.RWMutex + clients map[*client]struct{} + byUser map[string]map[*client]struct{} +} + +func NewHub(log *slog.Logger, originOK func(*http.Request) bool) *Hub { + return &Hub{ + log: log, + originOK: originOK, + clients: map[*client]struct{}{}, + byUser: map[string]map[*client]struct{}{}, + } +} + +// SetInbound installs the client-message handler (chat typing etc.). +func (h *Hub) SetInbound(f Inbound) { h.inbound = f } + +// ConnectedUserIDs lists users with at least one live socket. +func (h *Hub) ConnectedUserIDs() []string { + h.mu.RLock() + defer h.mu.RUnlock() + out := make([]string, 0, len(h.byUser)) + for id := range h.byUser { + out = append(out, id) + } + return out +} + +func (h *Hub) register(c *client) { + h.mu.Lock() + defer h.mu.Unlock() + h.clients[c] = struct{}{} + if h.byUser[c.userID] == nil { + h.byUser[c.userID] = map[*client]struct{}{} + } + h.byUser[c.userID][c] = struct{}{} +} + +func (h *Hub) unregister(c *client) { + h.mu.Lock() + defer h.mu.Unlock() + delete(h.clients, c) + if set := h.byUser[c.userID]; set != nil { + delete(set, c) + if len(set) == 0 { + delete(h.byUser, c.userID) + } + } +} + +func encode(channel, event string, payload any) []byte { + data, err := json.Marshal(payload) + if err != nil { + data = []byte("null") + } + b, _ := json.Marshal(Message{Channel: channel, Event: event, Data: data}) + return b +} + +// Broadcast sends to every connected client. +func (h *Hub) Broadcast(channel, event string, payload any) { + b := encode(channel, event, payload) + h.mu.RLock() + defer h.mu.RUnlock() + for c := range h.clients { + select { + case c.send <- b: + default: // slow client: drop rather than block the publisher + } + } +} + +// SendTo delivers to all sockets of the given users. +func (h *Hub) SendTo(userIDs []string, channel, event string, payload any) { + b := encode(channel, event, payload) + h.mu.RLock() + defer h.mu.RUnlock() + for _, id := range userIDs { + for c := range h.byUser[id] { + select { + case c.send <- b: + default: + } + } + } +} + +// Handle upgrades the request (the caller has already authenticated userID). +func (h *Hub) Handle(w http.ResponseWriter, r *http.Request, userID string) { + if h.originOK != nil && !h.originOK(r) { + http.Error(w, "origin not allowed", http.StatusForbidden) + return + } + conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{ + // Origin verified above against APP_BASE_URL (§12). + InsecureSkipVerify: true, + }) + if err != nil { + h.log.Warn("ws accept", "err", err) + return + } + c := &client{userID: userID, send: make(chan []byte, 64)} + h.register(c) + defer h.unregister(c) + + ctx, cancel := context.WithCancel(r.Context()) + defer cancel() + go h.writeLoop(ctx, cancel, conn, c) + h.readLoop(ctx, conn, c) + conn.Close(websocket.StatusNormalClosure, "bye") +} + +func (h *Hub) writeLoop(ctx context.Context, cancel func(), conn *websocket.Conn, c *client) { + defer cancel() + heartbeat := time.NewTicker(30 * time.Second) + defer heartbeat.Stop() + for { + select { + case <-ctx.Done(): + return + case msg := <-c.send: + wctx, wcancel := context.WithTimeout(ctx, 10*time.Second) + err := conn.Write(wctx, websocket.MessageText, msg) + wcancel() + if err != nil { + return + } + case <-heartbeat.C: + pctx, pcancel := context.WithTimeout(ctx, 10*time.Second) + err := conn.Ping(pctx) + pcancel() + if err != nil { + return + } + } + } +} + +func (h *Hub) readLoop(ctx context.Context, conn *websocket.Conn, c *client) { + for { + _, data, err := conn.Read(ctx) + if err != nil { + return + } + if h.inbound == nil { + continue + } + var msg Message + if err := json.Unmarshal(data, &msg); err != nil { + continue + } + h.inbound(c.userID, msg) + } +} diff --git a/web/static/css/app.css b/web/static/css/app.css index 18c786c..bcdeaff 100644 --- a/web/static/css/app.css +++ b/web/static/css/app.css @@ -155,6 +155,23 @@ table.list th { color: var(--muted); font-size: 0.875rem; } .toast.err { border-left-color: var(--err); } .toast.ok { border-left-color: var(--ok); } +/* atomizing progress shimmer */ +.shimmer { + position: relative; + overflow: hidden; +} +.shimmer::after { + content: ""; + position: absolute; inset: 0; + background: linear-gradient(90deg, transparent, color-mix(in srgb, var(--accent) 12%, transparent), transparent); + animation: shimmer 1.6s infinite; +} +@keyframes shimmer { + from { transform: translateX(-100%); } + to { transform: translateX(100%); } +} +input[type=range] { width: 100%; accent-color: var(--accent); } + .tabs { display: flex; gap: 8px; margin-bottom: 16px; flex-wrap: wrap; } .tabs [aria-selected=true] { background: var(--accent); color: var(--accent-contrast); border-color: var(--accent); } diff --git a/web/static/js/atomization.js b/web/static/js/atomization.js new file mode 100644 index 0000000..5b27220 --- /dev/null +++ b/web/static/js/atomization.js @@ -0,0 +1,319 @@ +import { api, toast } from '/static/js/api.js'; +import { subscribe, onPollFallback } from '/static/js/ws.js'; + +const errorBox = document.getElementById('error'); +const board = document.getElementById('board'); +const customerSel = document.getElementById('board-customer'); +const icons = { epic: '◆', story: '▣', task: '▢' }; + +let customers = []; +let tasks = []; +let atomizerUp = true; +const selected = new Set(); + +function esc(s) { + return String(s ?? '').replace(/[&<>"']/g, (c) => ({ + '&': '&', '<': '<', '>': '>', '"': '"', "'": ''', + }[c])); +} + +async function checkHealth() { + const badge = document.getElementById('atomizer-health'); + try { + const res = await api('GET', '/api/v1/service-health'); + const a = res.atomizer; + atomizerUp = a.healthy && a.breaker !== 'open'; + badge.textContent = atomizerUp + ? `atomizer healthy · ${a.latencyMs} ms` + : `atomizer ${a.healthy ? 'breaker ' + a.breaker : 'DOWN'} — Subdivide/Extend disabled`; + badge.style.color = atomizerUp ? 'var(--ok)' : 'var(--err)'; + } catch (e) { + atomizerUp = false; + badge.textContent = 'atomizer status unknown'; + } + document.querySelectorAll('[data-needs-atomizer]').forEach((b) => { + b.disabled = !atomizerUp; + b.title = atomizerUp ? '' : 'Atomization Service is unavailable'; + }); +} + +async function load() { + try { + const cid = customerSel.value || ''; + const res = await api('GET', `/api/v1/consultant/board?customerId=${encodeURIComponent(cid)}`); + customers = res.customers; + tasks = res.tasks; + if (!customerSel.options.length) { + customerSel.replaceChildren( + new Option('All customers', ''), + ...customers.map((c) => new Option(c.name, c.id)), + ); + customerSel.value = new URLSearchParams(location.search).get('customerId') || ''; + if (customerSel.value) return load(); + } + render(); + } catch (e) { errorBox.textContent = e.message; } +} + +function childrenOf(id) { return tasks.filter((t) => t.parentId === id); } + +function render() { + selected.forEach((id) => { if (!tasks.find((t) => t.id === id)) selected.delete(id); }); + updateBulkButton(); + const roots = tasks.filter((t) => !t.parentId); + board.replaceChildren(...roots.map(renderRoot)); + if (!roots.length) { + board.innerHTML = '

No tasks on the atomization board for this selection.

'; + } + checkHealth(); +} + +function statusBadge(t) { + const cls = { atomizing: 'warn', published: 'ok' }[t.status] || ''; + return `${esc(t.status)}`; +} + +function renderRoot(root) { + const card = document.createElement('div'); + card.className = 'card' + (root.status === 'atomizing' ? ' shimmer' : ''); + const cust = customers.find((c) => c.id === root.customerId); + const icon = root.external ? (icons[root.external.type] || '▢') : '▢'; + const orphan = root.external && root.external.orphaned + ? 'orphaned' : ''; + card.innerHTML = ` +
+

${icon} ${esc(root.title)} ${statusBadge(root)} ${orphan}

+ ${esc(cust ? cust.name : '')} ${root.external ? '· ' + esc(root.external.key) : ''} +
+

${esc((root.description || '').slice(0, 240))}

+
+ Budget ${root.budget} · bounty preview ${root.bounty} + + ${root.external && root.external.url ? `source ↗` : ''} + + + + + +
+
`; + card.querySelector('[data-act=subdivide]').addEventListener('click', () => openSubdivide(root)); + card.querySelector('[data-act=extend]').addEventListener('click', () => openExtend(root)); + card.querySelector('[data-act=edit]').addEventListener('click', () => openEdit(root, true)); + card.querySelector('[data-act=archive]').addEventListener('click', () => archive(root)); + + const kids = childrenOf(root.id); + if (kids.length) { + card.querySelector('.children').appendChild(renderChildren(root, kids, 1)); + } + return card; +} + +function renderChildren(parent, kids, depth) { + const wrap = document.createElement('div'); + wrap.style.marginLeft = `${depth * 16}px`; + wrap.className = 'stack mt'; + + const subdivided = kids.filter((k) => k.origin === 'subdivided'); + const sum = subdivided.reduce((acc, k) => acc + k.effortCoefficient, 0); + const hasExtension = kids.some((k) => k.origin === 'extended'); + if (subdivided.length) { + const ind = document.createElement('p'); + const bad = !hasExtension && Math.abs(sum - 1) > 0.005; + ind.innerHTML = `Coefficient sum: ${sum.toFixed(2)}` + + (hasExtension ? ' (extensions allow > 1.00)' + : bad ? ' — should be 1.00' : ''); + wrap.appendChild(ind); + } + + kids.forEach((k) => { + const row = document.createElement('div'); + row.className = 'card' + (k.status === 'atomizing' ? ' shimmer' : ''); + row.style.padding = '12px 16px'; + const ext = k.origin === 'extended' ? 'extension' : ''; + row.innerHTML = ` +
+ + ${k.status === 'atomized' ? `` : ''} + ${esc(k.title)} ${ext} ${statusBadge(k)} + + bounty ${k.bounty} +
+

${esc((k.description || '').slice(0, 160))}

+
+ + + + + ${k.status === 'atomized' ? '' : ''} + + +
`; + const slider = row.querySelector('input[type=range]'); + let sliderTimer = null; + slider.addEventListener('input', () => { + row.querySelector('[data-coeff]').textContent = parseFloat(slider.value).toFixed(2); + row.querySelector('[data-bounty]').textContent = (slider.value * k.budget).toFixed(2); + clearTimeout(sliderTimer); + sliderTimer = setTimeout(async () => { + try { + const res = await api('PATCH', `/api/v1/tasks/${k.id}`, { + effortCoefficient: parseFloat(slider.value), version: k.version, + }); + k.version = res.task.version; + k.effortCoefficient = res.task.effortCoefficient; + k.bounty = res.task.bounty; + render(); + } catch (e) { + if (e.status === 409) { toast('Task changed elsewhere — reloading.', 'err'); load(); } + else toast(e.message, 'err'); + } + }, 400); + }); + const sel = row.querySelector('[data-sel]'); + if (sel) { + sel.addEventListener('change', () => { + if (sel.checked) selected.add(k.id); else selected.delete(k.id); + updateBulkButton(); + }); + } + row.querySelector('[data-act=edit]').addEventListener('click', () => openEdit(k, false)); + row.querySelector('[data-act=extend]').addEventListener('click', () => openExtend(k)); + row.querySelector('[data-act=archive]').addEventListener('click', () => archive(k)); + const pub = row.querySelector('[data-act=publish]'); + if (pub) { + pub.addEventListener('click', async () => { + try { await api('POST', `/api/v1/tasks/${k.id}/publish`, {}); toast('Published.', 'ok'); load(); } + catch (e) { toast(e.message, 'err'); } + }); + } + wrap.appendChild(row); + const grand = childrenOf(k.id); + if (grand.length) wrap.appendChild(renderChildren(k, grand, depth + 1)); + }); + return wrap; +} + +async function archive(t) { + if (!confirm(`Archive "${t.title}"?`)) return; + try { await api('POST', `/api/v1/tasks/${t.id}/archive`, {}); load(); } + catch (e) { toast(e.message, 'err'); } +} + +function updateBulkButton() { + const btn = document.getElementById('bulk-publish'); + btn.textContent = `Publish selected (${selected.size})`; + btn.disabled = selected.size === 0; +} + +document.getElementById('select-all').addEventListener('click', () => { + tasks.filter((t) => t.status === 'atomized').forEach((t) => selected.add(t.id)); + render(); +}); + +document.getElementById('bulk-publish').addEventListener('click', async () => { + try { + const res = await api('POST', '/api/v1/tasks/publish', { ids: [...selected] }); + toast(`Published ${res.published.length} task(s)` + + (Object.keys(res.failed).length ? `, ${Object.keys(res.failed).length} failed` : '.'), + Object.keys(res.failed).length ? 'err' : 'ok'); + selected.clear(); + load(); + } catch (e) { toast(e.message, 'err'); } +}); + +// ---- dialogs ---- +let dialogTask = null; + +const sdDlg = document.getElementById('subdivide-dialog'); +function openSubdivide(t) { + dialogTask = t; + document.getElementById('subdivide-task-title').textContent = t.title; + document.getElementById('sd-replace-warning').hidden = !childrenOf(t.id).length; + sdDlg.showModal(); +} +document.getElementById('sd-cancel').addEventListener('click', () => sdDlg.close()); +document.getElementById('subdivide-form').addEventListener('submit', async (e) => { + e.preventDefault(); + try { + await api('POST', `/api/v1/tasks/${dialogTask.id}/subdivide`, { + note: document.getElementById('sd-note').value.trim(), + constraints: { + minTasks: parseInt(document.getElementById('sd-min').value, 10) || 0, + maxTasks: parseInt(document.getElementById('sd-max').value, 10) || 0, + }, + confirmReplace: !document.getElementById('sd-replace-warning').hidden, + }); + sdDlg.close(); + toast('Subdivision queued — results stream in shortly.', 'ok'); + load(); + } catch (err) { toast(err.message, 'err'); } +}); + +const exDlg = document.getElementById('extend-dialog'); +function openExtend(t) { + dialogTask = t; + document.getElementById('extend-task-title').textContent = t.title; + exDlg.showModal(); +} +document.getElementById('ex-cancel').addEventListener('click', () => exDlg.close()); +document.getElementById('extend-form').addEventListener('submit', async (e) => { + e.preventDefault(); + try { + await api('POST', `/api/v1/tasks/${dialogTask.id}/extend`, { + note: document.getElementById('ex-note').value.trim(), + }); + exDlg.close(); + document.getElementById('ex-note').value = ''; + toast('Extension queued.', 'ok'); + } catch (err) { toast(err.message, 'err'); } +}); + +const edDlg = document.getElementById('edit-dialog'); +function openEdit(t, isRoot) { + dialogTask = t; + document.getElementById('ed-title').value = t.title; + document.getElementById('ed-desc').value = t.description || ''; + document.getElementById('ed-ac').value = (t.acceptanceCriteria || []).join('\n'); + document.getElementById('ed-budget').value = t.budget; + document.getElementById('ed-cascade-wrap').style.display = isRoot ? '' : 'none'; + edDlg.showModal(); +} +document.getElementById('ed-cancel').addEventListener('click', () => edDlg.close()); +document.getElementById('edit-form').addEventListener('submit', async (e) => { + e.preventDefault(); + try { + await api('PATCH', `/api/v1/tasks/${dialogTask.id}`, { + version: dialogTask.version, + title: document.getElementById('ed-title').value.trim(), + description: document.getElementById('ed-desc').value, + acceptanceCriteria: document.getElementById('ed-ac').value.split('\n').map((s) => s.trim()).filter(Boolean), + budget: parseFloat(document.getElementById('ed-budget').value) || 0, + cascadeBudget: document.getElementById('ed-cascade').checked && + document.getElementById('ed-cascade-wrap').style.display !== 'none', + }); + edDlg.close(); + toast('Saved.', 'ok'); + load(); + } catch (err) { + if (err.status === 409) { toast('Task changed elsewhere — reloading.', 'err'); load(); } + else toast(err.message, 'err'); + } +}); + +// ---- live updates ---- +let reloadTimer = null; +function scheduleReload() { + clearTimeout(reloadTimer); + reloadTimer = setTimeout(load, 300); +} +subscribe('board', () => scheduleReload()); +onPollFallback(load); +customerSel.addEventListener('change', load); +setInterval(checkHealth, 30000); +load(); diff --git a/web/static/js/ws.js b/web/static/js/ws.js new file mode 100644 index 0000000..502bfea --- /dev/null +++ b/web/static/js/ws.js @@ -0,0 +1,57 @@ +// Live updates: one multiplexed WebSocket with auto-reconnect and a 15 s +// polling fallback signal (§2.2). Subscribers register per channel. +const subs = new Map(); // channel -> Set +let socket = null; +let attempts = 0; +let fallbackPolling = false; +const pollListeners = new Set(); + +export function subscribe(channel, fn) { + if (!subs.has(channel)) subs.set(channel, new Set()); + subs.get(channel).add(fn); + ensureSocket(); + return () => subs.get(channel).delete(fn); +} + +// onPollFallback registers a callback invoked every 15 s while the socket is +// down, so pages can refresh their data without WS. +export function onPollFallback(fn) { + pollListeners.add(fn); +} + +export function send(channel, event, data) { + if (socket && socket.readyState === WebSocket.OPEN) { + socket.send(JSON.stringify({ channel, event, data })); + } +} + +let pollTimer = null; +function startPolling() { + if (fallbackPolling) return; + fallbackPolling = true; + pollTimer = setInterval(() => pollListeners.forEach((fn) => fn()), 15000); +} +function stopPolling() { + fallbackPolling = false; + if (pollTimer) clearInterval(pollTimer); +} + +function ensureSocket() { + if (socket && socket.readyState <= WebSocket.OPEN) return; + const proto = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; + socket = new WebSocket(`${proto}//${window.location.host}/ws`); + socket.addEventListener('open', () => { attempts = 0; stopPolling(); }); + socket.addEventListener('message', (e) => { + try { + const msg = JSON.parse(e.data); + (subs.get(msg.channel) || []).forEach((fn) => fn(msg.event, msg.data)); + } catch (err) { /* ignore malformed frames */ } + }); + socket.addEventListener('close', () => { + startPolling(); + attempts += 1; + const delay = Math.min(30000, 1000 * 2 ** attempts); + setTimeout(ensureSocket, delay); + }); + socket.addEventListener('error', () => socket.close()); +} diff --git a/web/templates/atomization.html b/web/templates/atomization.html new file mode 100644 index 0000000..23d1672 --- /dev/null +++ b/web/templates/atomization.html @@ -0,0 +1,91 @@ +{{define "content"}} +
+

Atomization Board

+ checking atomizer… +
+ + +
+ + + + + +
+ +
+ + +
+

Subdivide task

+

+
+ + +
+
+
+ + +
+
+ + +
+
+ +
+ + +
+
+ + + +
+

Extend task

+

+
+ + +
+
+ + +
+
+ + + +
+

Edit task

+
+ + +
+
+ + +
+
+ + +
+
+
+ + +
+
+ +
+
+
+ + +
+
+ +{{end}}