BountyBoard/internal/atomize/service.go

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