feat(stovepipe): Implement project processing

stovepipe/controller/projectanalysis/BUILD.bazel

@@ -1,18 +1,54 @@
-load("@rules_go//go:def.bzl", "go_library")
+load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
     srcs = ["projectanalysis.go"],
     importpath = "github.com/uber/submitqueue/stovepipe/controller/projectanalysis",
     visibility = ["//visibility:public"],
     deps = [
+        "//platform/base/id:go_default_library",
+        "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
+        "//platform/metrics:go_default_library",
+        "//platform/publish:go_default_library",
+        "//stovepipe/core/loader:go_default_library",
+        "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/core/projectissues:go_default_library",
         "//stovepipe/core/requestlog:go_default_library",
+        "//stovepipe/core/validationfact:go_default_library",
+        "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/projectissues:go_default_library",
         "//stovepipe/extension/projectstatus:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
         "@org_uber_go_zap//:go_default_library",
     ],
 )
+
+go_test(
+    name = "go_default_test",
+    srcs = ["projectanalysis_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//platform/base/messagequeue:go_default_library",
+        "//platform/consumer:go_default_library",
+        "//platform/consumer/mock:go_default_library",
+        "//platform/extension/messagequeue/mock:go_default_library",
+        "//platform/metrics:go_default_library",
+        "//stovepipe/core/messagequeue:go_default_library",
+        "//stovepipe/core/requestlog:go_default_library",
+        "//stovepipe/core/requestlog/mock:go_default_library",
+        "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/projectissues:go_default_library",
+        "//stovepipe/extension/projectissues/mock:go_default_library",
+        "//stovepipe/extension/projectstatus:go_default_library",
+        "//stovepipe/extension/projectstatus/mock:go_default_library",
+        "//stovepipe/extension/storage:go_default_library",
+        "//stovepipe/extension/storage/mock:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+    ],
+)

stovepipe/controller/projectanalysis/README.md

@@ -0,0 +1,12 @@
+# Project analysis
+
+Project analysis consumes request IDs after terminal build analysis has retained
+all final build issues. A missing terminal-analysis completion is an invalid
+handoff, not a reason to poll.
+
+The controller resolves project statuses and records their facts and lookup entries before attributing issues. It materializes both directions of each issue/project association. Each step retains its completion event after its writes; repository recording is queued only after both steps finish.
+
+Retries adopt immutable facts and repair missing lookup entries. A completed
+request skips provider calls and retries only the recording handoff. Empty
+results still record completion. Cancelled requests bypass project analysis;
+superseded requests require no work.

stovepipe/controller/projectanalysis/projectanalysis.go

@@ -12,14 +12,29 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-// Package projectanalysis owns terminal project processing.
+// Package projectanalysis records project statuses and issue attribution after build analysis.
 package projectanalysis
 
 import (
+	"context"
+	"errors"
+	"fmt"
+	"math"
+	"strconv"
+	"time"
+
 	"github.com/uber-go/tally"
+	"github.com/uber/submitqueue/platform/base/id"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
 	"github.com/uber/submitqueue/platform/consumer"
+	"github.com/uber/submitqueue/platform/metrics"
+	"github.com/uber/submitqueue/platform/publish"
+	"github.com/uber/submitqueue/stovepipe/core/loader"
+	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	projectissuescore "github.com/uber/submitqueue/stovepipe/core/projectissues"
 	"github.com/uber/submitqueue/stovepipe/core/requestlog"
+	"github.com/uber/submitqueue/stovepipe/core/validationfact"
+	"github.com/uber/submitqueue/stovepipe/entity"
 	"github.com/uber/submitqueue/stovepipe/extension/projectissues"
 	"github.com/uber/submitqueue/stovepipe/extension/projectstatus"
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
@@ -64,6 +79,194 @@ func NewController(
 	}
 }
 
+func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
+	op := metrics.Begin(c.scope, "process", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...)
+	defer func() { op.Complete(retErr) }()
+
+	msg := delivery.Message()
+	analysis := &stovepipemq.ProjectAnalysis{}
+	if err := stovepipemq.Unmarshal(msg.Payload, analysis); err != nil {
+		return fmt.Errorf("deserialize project analysis: %w", err)
+	}
+	if err := entityqueue.ValidatePayloadQueue(msg, analysis.GetQueueName()); err != nil {
+		return fmt.Errorf("invalid project analysis identity: %w", err)
+	}
+	if err := id.Validate(analysis.GetId()); err != nil {
+		return fmt.Errorf("invalid project analysis request ID: %w", err)
+	}
+	if msg.PartitionKey != analysis.GetId() {
+		return fmt.Errorf("project analysis must be partitioned by request ID")
+	}
+	store, err := c.stores.For(storage.Config{QueueName: analysis.GetQueueName()})
+	if err != nil {
+		return fmt.Errorf("resolve project analysis storage: %w", err)
+	}
+	request, err := loader.ByID(ctx, analysis.GetId(), store.GetRequestStore().Get, "project analysis request")
+	if err != nil {
+		return err
+	}
+	if request.Queue != analysis.GetQueueName() {
+		return fmt.Errorf("project analysis request belongs to another queue")
+	}
+	switch request.State {
+	case entity.RequestStateSucceeded, entity.RequestStateFailed:
+	case entity.RequestStateCancelled:
+		return publishRecord(ctx, c.registry, request)
+	case entity.RequestStateSuperseded:
+		return nil
+	default:
+		return fmt.Errorf("project analysis requires a terminal request")
+	}
+	if request.TerminalBuildID == "" {
+		return fmt.Errorf("project analysis requires the request's terminal build")
+	}
+
+	projectIssuesRecordedLog := requestlog.NewRequestEventLog(request, entity.RequestEventProjectIssuesRecorded, request.TerminalBuildID, nil)
+	history := store.GetRequestLogStore()
+	_, err = history.Get(ctx, request.ID, projectIssuesRecordedLog.ID)
+	if err == nil {
+		return publishRecord(ctx, c.registry, request)
+	}
+	if !errors.Is(err, storage.ErrNotFound) {
+		return fmt.Errorf("read project analysis completion: %w", err)
+	}
+	buildAnalysisCompletedLog := requestlog.NewRequestEventLog(request, entity.RequestEventBuildAnalysisCompleted, request.TerminalBuildID, nil)
+	if _, err := history.Get(ctx, request.ID, buildAnalysisCompletedLog.ID); err != nil {
+		return fmt.Errorf("require completed terminal build analysis: %w", err)
+	}
+
+	if err := c.recordProjectStatuses(ctx, store, request); err != nil {
+		return err
+	}
+	if err := c.recordIssueAttribution(ctx, store, request); err != nil {
+		return err
+	}
+	return publishRecord(ctx, c.registry, request)
+}
+
+func (c *Controller) recordProjectStatuses(ctx context.Context, store storage.Storage, request entity.Request) (retErr error) {
+	op := metrics.Begin(c.scope, "record_project_statuses", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...)
+	defer func() { op.Complete(retErr) }()
+
+	resolver, err := c.resolvers.For(projectstatus.ResolverConfig{QueueName: request.Queue})
+	if err != nil {
+		return fmt.Errorf("resolve project status resolver for queue %q: %w", request.Queue, err)
+	}
+	projectStatuses, err := resolver.Resolve(ctx, request)
+	if err != nil {
+		return fmt.Errorf("resolve project statuses for request %q: %w", request.ID, err)
+	}
+
+	if err := validateProjectStatuses(request, projectStatuses); err != nil {
+		return err
+	}
+	metrics.NamedHistogram(c.scope, "record_project_statuses", "projects", metrics.CountBuckets,
+		metrics.TagsFromContext(ctx)...,
+	).RecordValue(float64(len(projectStatuses)))
+
+	brokenCount := 0
+	for _, status := range projectStatuses {
+		fact, _, err := validationfact.Persist(ctx, c.logger, c.scope, store.GetValidationFactStore(), entity.ValidationFact{
+			URI:         request.URI,
+			Project:     status.ProjectInfo.Name,
+			ProjectType: status.ProjectInfo.ProjectType,
+			Degree:      status.Degree,
+			RequestID:   request.ID,
+			CreatedAt:   time.Now().UnixMilli(),
+		})
+		if err != nil {
+			return err
+		}
+		if !fact.IsGreen() {
+			brokenCount++
+		}
+
+		entry, err := entity.NewProjectFactByDegree(fact)
+		if err != nil {
+			return err
+		}
+
+		if err := store.GetProjectFactByDegreeStore().Create(ctx, entry); err != nil && !errors.Is(err, storage.ErrAlreadyExists) {
+			return fmt.Errorf("failed to index project fact for request %q and project %q: %w", request.ID, fact.Project, err)
+		}
+	}
+	projectFactsRecordedLog := requestlog.NewRequestEventLog(
+		request,
+		entity.RequestEventProjectFactsRecorded,
+		"all",
+		map[string]string{
+			requestlog.MetadataKeyProjectFactCount:       strconv.Itoa(len(projectStatuses)),
+			requestlog.MetadataKeyBrokenProjectFactCount: strconv.Itoa(brokenCount),
+		},
+	)
+	if err := c.requestLogMaterializer.PersistLog(ctx, store, projectFactsRecordedLog); err != nil {
+		return fmt.Errorf("failed to record project facts for request %s: %w", request.ID, err)
+	}
+	return nil
+}
+
+func (c *Controller) recordIssueAttribution(ctx context.Context, store storage.Storage, request entity.Request) error {
+	attributer, err := c.attributers.For(projectissues.AttributerConfig{QueueName: request.Queue})
+	if err != nil {
+		return fmt.Errorf("resolve project issue attributer: %w", err)
+	}
+	attributions, err := attributer.Attribute(ctx, request)
+	if err != nil {
+		return fmt.Errorf("attribute issues for request %q: %w", request.ID, err)
+	}
+	for _, attribution := range attributions {
+		if attribution.TaskID == "" {
+			return fmt.Errorf("project issue attribution requires a task ID")
+		}
+		if attribution.UnitID == "" {
+			return fmt.Errorf("project issue attribution requires a unit-level issue")
+		}
+		for _, project := range attribution.Projects {
+			entry := entity.ProjectIssueAttribution{TaskID: attribution.TaskID, UnitID: attribution.UnitID, Project: project}
+			if err := c.projectIssueMaterializer.PersistAttribution(ctx, store, request.ID, entry); err != nil {
+				return err
+			}
+		}
+	}
+	projectIssuesRecordedLog := requestlog.NewRequestEventLog(request, entity.RequestEventProjectIssuesRecorded, request.TerminalBuildID, nil)
+	if err := c.requestLogMaterializer.PersistLog(ctx, store, projectIssuesRecordedLog); err != nil {
+		return fmt.Errorf("record project issue completion: %w", err)
+	}
+	return nil
+}
+
+func validateProjectStatuses(request entity.Request, projectStatuses []projectstatus.Result) error {
+	seen := make(map[string]struct{}, len(projectStatuses))
+	for _, status := range projectStatuses {
+		if status.ProjectInfo.Name == "" {
+			return fmt.Errorf("project status for request %q has an empty project", request.ID)
+		}
+		if _, ok := seen[status.ProjectInfo.Name]; ok {
+			return fmt.Errorf("project status for request %q contains duplicate project %q", request.ID, status.ProjectInfo.Name)
+		}
+		seen[status.ProjectInfo.Name] = struct{}{}
+		if math.IsNaN(status.Degree) || status.Degree < entity.DegreeGreen || status.Degree > entity.DegreeBroken {
+			return fmt.Errorf("project status for request %q and project %q has invalid degree %v", request.ID, status.ProjectInfo.Name, status.Degree)
+		}
+	}
+	return nil
+}
+
+func publishRecord(ctx context.Context, registry consumer.TopicRegistry, request entity.Request) error {
+	payload, err := stovepipemq.Marshal(&stovepipemq.Record{Id: request.ID, QueueName: request.Queue})
+	if err != nil {
+		return fmt.Errorf("serialize repository recording request: %w", err)
+	}
+	return publish.Message(ctx, registry, stovepipemq.TopicKeyRecord, publish.MessageParams{
+		Tenant:       request.Queue,
+		ID:           publish.IntentID(request.ID),
+		Payload:      payload,
+		PartitionKey: request.ID,
+	})
+}
+
 func (c *Controller) Name() string                { return "projectanalysis" }
 func (c *Controller) TopicKey() consumer.TopicKey { return c.topicKey }
 func (c *Controller) ConsumerGroup() string       { return c.consumerGroup }
+
+var _ consumer.Controller = (*Controller)(nil)