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, CustomerID: t.CustomerID, } // Project context for backends that need it (the Anypreta `system` // object). Not worth failing an atomization over. if cust, err := s.st.CustomerByID(ctx, t.CustomerID); err == nil { req.CustomerName = cust.Name } else { s.log.Warn("subdivide: customer lookup failed, sending id only", "task", t.ID, "customer", t.CustomerID, "err", err) } 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) }