refactor(stovepipe): make IngestController operate on entities

service/stovepipe/server/BUILD.bazel

@@ -13,6 +13,7 @@ go_library(
         "//platform/errs/mysql:go_default_library",
         "//platform/extension/messagequeue:go_default_library",
         "//platform/extension/messagequeue/mysql:go_default_library",
+        "//service/stovepipe/server/mapper:go_default_library",
         "//stovepipe/controller:go_default_library",
         "//stovepipe/controller/process:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",

service/stovepipe/server/main.go

@@ -35,6 +35,7 @@ import (
 	mysqlerrs "github.com/uber/submitqueue/platform/errs/mysql"
 	extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
 	queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql"
+	"github.com/uber/submitqueue/service/stovepipe/server/mapper"
 	"github.com/uber/submitqueue/stovepipe/controller"
 	"github.com/uber/submitqueue/stovepipe/controller/process"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
@@ -59,9 +60,14 @@ func (s *StovepipeServer) Ping(ctx context.Context, req *pb.PingRequest) (*pb.Pi
 	return s.pingController.Ping(ctx, req)
 }
 
-// Ingest delegates to the controller.
+// Ingest maps the wire request to an entity, delegates to the controller, and maps
+// the result back to the wire response.
 func (s *StovepipeServer) Ingest(ctx context.Context, req *pb.IngestRequest) (*pb.IngestResponse, error) {
-	return s.ingestController.Ingest(ctx, req)
+	result, err := s.ingestController.Ingest(ctx, mapper.ProtoToIngestRequest(req))
+	if err != nil {
+		return nil, err
+	}
+	return mapper.IngestResultToProto(result), nil
 }
 
 // inMemoryCounter is a minimal, process-local counter.Counter used to wire the example

service/stovepipe/server/mapper/BUILD.bazel

@@ -0,0 +1,23 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["ingest.go"],
+    importpath = "github.com/uber/submitqueue/service/stovepipe/server/mapper",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//api/stovepipe/protopb:go_default_library",
+        "//stovepipe/entity:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["ingest_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//api/stovepipe/protopb:go_default_library",
+        "//stovepipe/entity:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+    ],
+)

service/stovepipe/server/mapper/ingest.go

@@ -0,0 +1,38 @@
+// Copyright (c) 2025 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package mapper translates stovepipe wire (proto) types to and from the domain
+// entities the controllers operate on. Each RPC gets its own file; translation
+// lives here so controllers stay proto-free.
+package mapper
+
+import (
+	pb "github.com/uber/submitqueue/api/stovepipe/protopb"
+	"github.com/uber/submitqueue/stovepipe/entity"
+)
+
+// ProtoToIngestRequest maps the wire IngestRequest to the entity.IngestRequest
+// the controller operates on.
+func ProtoToIngestRequest(req *pb.IngestRequest) entity.IngestRequest {
+	return entity.IngestRequest{
+		Queue: req.GetQueue(),
+	}
+}
+
+// IngestResultToProto maps the domain result to the wire IngestResponse.
+func IngestResultToProto(result entity.IngestResult) *pb.IngestResponse {
+	return &pb.IngestResponse{
+		Id: result.ID,
+	}
+}

service/stovepipe/server/mapper/ingest_test.go

@@ -0,0 +1,75 @@
+// Copyright (c) 2025 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mapper
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	pb "github.com/uber/submitqueue/api/stovepipe/protopb"
+	"github.com/uber/submitqueue/stovepipe/entity"
+)
+
+func TestProtoToIngestRequest(t *testing.T) {
+	tests := []struct {
+		name     string
+		req      *pb.IngestRequest
+		expected entity.IngestRequest
+	}{
+		{
+			name:     "maps queue",
+			req:      &pb.IngestRequest{Queue: "monorepo/main"},
+			expected: entity.IngestRequest{Queue: "monorepo/main"},
+		},
+		{
+			name:     "empty request yields zero value",
+			req:      &pb.IngestRequest{},
+			expected: entity.IngestRequest{},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := ProtoToIngestRequest(tt.req)
+			assert.Equal(t, tt.expected, got)
+		})
+	}
+}
+
+func TestIngestResultToProto(t *testing.T) {
+	tests := []struct {
+		name     string
+		result   entity.IngestResult
+		expected *pb.IngestResponse
+	}{
+		{
+			name:     "maps ID",
+			result:   entity.IngestResult{ID: "request/monorepo/main/7"},
+			expected: &pb.IngestResponse{Id: "request/monorepo/main/7"},
+		},
+		{
+			name:     "zero value result yields empty response",
+			result:   entity.IngestResult{},
+			expected: &pb.IngestResponse{},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := IngestResultToProto(tt.result)
+			assert.Equal(t, tt.expected, got)
+		})
+	}
+}

stovepipe/controller/ingest.go

@@ -20,7 +20,6 @@ import (
 	"fmt"
 
 	"github.com/uber-go/tally"
-	pb "github.com/uber/submitqueue/api/stovepipe/protopb"
 	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
 	"github.com/uber/submitqueue/platform/consumer"
 	"github.com/uber/submitqueue/platform/errs"
@@ -87,47 +86,47 @@ func NewIngestController(
 // URI mapping committed but the request write failed — completes the missing steps instead of
 // returning a dangling reference. The (queue, URI) mapping is the dedup gate, so concurrent
 // ingests of the same head converge on one request.
-func (c *IngestController) Ingest(ctx context.Context, req *pb.IngestRequest) (resp *pb.IngestResponse, retErr error) {
+func (c *IngestController) Ingest(ctx context.Context, req entity.IngestRequest) (result entity.IngestResult, retErr error) {
 	const opName = "ingest"
 
 	op := metrics.Begin(c.metricsScope, opName)
 	defer func() { op.Complete(retErr) }()
 
 	if req.Queue == "" {
-		return nil, fmt.Errorf("IngestController requires the request to have a queue name specified: %w", ErrInvalidRequest)
+		return entity.IngestResult{}, fmt.Errorf("IngestController requires the request to have a queue name specified: %w", ErrInvalidRequest)
 	}
 	queue := req.Queue
 
 	// Resolve the queue's current head commit to its opaque URI via SourceControl.
 	// An unresolvable queue/ref is a caller error (unknown queue), not infrastructure.
 	sc, err := c.sourceControl.For(sourcecontrol.Config{QueueName: queue})
 	if err != nil {
-		return nil, fmt.Errorf("IngestController failed to resolve source control for queue=%s: %w", queue, err)
+		return entity.IngestResult{}, fmt.Errorf("IngestController failed to resolve source control for queue=%s: %w", queue, err)
 	}
 	uri, err := sc.Latest(ctx)
 	if err != nil {
 		if sourcecontrol.IsNotFound(err) {
-			return nil, fmt.Errorf("IngestController could not resolve head for queue=%s: %w: %w", queue, err, ErrInvalidRequest)
+			return entity.IngestResult{}, fmt.Errorf("IngestController could not resolve head for queue=%s: %w: %w", queue, err, ErrInvalidRequest)
 		}
-		return nil, fmt.Errorf("IngestController failed to resolve head for queue=%s: %w", queue, err)
+		return entity.IngestResult{}, fmt.Errorf("IngestController failed to resolve head for queue=%s: %w", queue, err)
 	}
 
 	// The (queue, URI) mapping is the dedup gate and the source of truth for "does this head
 	// have a request id".
 	id, err := c.resolveID(ctx, queue, uri)
 	if err != nil {
-		return nil, err
+		return entity.IngestResult{}, err
 	}
 
 	// Ensure the request row exists, healing a prior partial write where the mapping committed
 	// but the request did not.
 	request, err := c.ensureRequest(ctx, id, queue, uri)
 	if err != nil {
-		return nil, err
+		return entity.IngestResult{}, err
 	}
 
 	if err := c.advanceQueueLatestRequestID(ctx, queue, id); err != nil {
-		return nil, err
+		return entity.IngestResult{}, err
 	}
 
 	// Publish while the request is still pre-pipeline (Accepted). The process consumer is
@@ -136,7 +135,7 @@ func (c *IngestController) Ingest(ctx context.Context, req *pb.IngestRequest) (r
 	// process advances the request past Accepted, ingest stops re-publishing.
 	if request.State == entity.RequestStateAccepted {
 		if err := c.publishProcess(ctx, id, queue); err != nil {
-			return nil, fmt.Errorf("IngestController failed to publish request %s to process: %w", id, err)
+			return entity.IngestResult{}, fmt.Errorf("IngestController failed to publish request %s to process: %w", id, err)
 		}
 	}
 
@@ -147,7 +146,7 @@ func (c *IngestController) Ingest(ctx context.Context, req *pb.IngestRequest) (r
 		"state", request.State,
 	)
 
-	return &pb.IngestResponse{Id: id}, nil
+	return entity.IngestResult{ID: id}, nil
 }
 
 // resolveID returns the request id mapped to (queue, URI), minting and claiming a new one if the

stovepipe/controller/ingest_test.go

@@ -22,7 +22,6 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/uber-go/tally"
-	pb "github.com/uber/submitqueue/api/stovepipe/protopb"
 	"github.com/uber/submitqueue/platform/consumer"
 	countermock "github.com/uber/submitqueue/platform/extension/counter/mock"
 	mqmock "github.com/uber/submitqueue/platform/extension/messagequeue/mock"
@@ -145,7 +144,6 @@ func TestIngestController_Ingest(t *testing.T) {
 			name:  "heals when uri mapped but request missing",
 			queue: testQueue,
 			setup: func(m ingestMocks) {
-				// Prior attempt committed the URI mapping but failed before the request write.
 				expectResolve(m)
 				m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testQueue, testURI).Return("request/monorepo/main/3", nil)
 				m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(entity.Request{}, storage.ErrNotFound)
@@ -242,15 +240,15 @@ func TestIngestController_Ingest(t *testing.T) {
 			c, m := newIngestController(t, ctrl)
 			tt.setup(m)
 
-			resp, err := c.Ingest(context.Background(), &pb.IngestRequest{Queue: tt.queue})
+			result, err := c.Ingest(context.Background(), entity.IngestRequest{Queue: tt.queue})
 
 			if tt.wantErr {
 				require.Error(t, err)
 				assert.Equal(t, tt.wantInvalid, IsInvalidRequest(err))
 				return
 			}
 			require.NoError(t, err)
-			assert.Equal(t, tt.wantID, resp.Id)
+			assert.Equal(t, tt.wantID, result.ID)
 		})
 	}
 }

stovepipe/entity/BUILD.bazel

@@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "go_default_library",
     srcs = [
+        "ingest.go",
         "queue.go",
         "queue_config.go",
         "request.go",

stovepipe/entity/ingest.go

@@ -0,0 +1,29 @@
+// Copyright (c) 2025 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package entity
+
+// IngestRequest represents the validated inputs of an ingest RPC call.
+// The controller resolves the queue's head commit and mints a request ID internally.
+type IngestRequest struct {
+	// Queue is the name of the queue whose head commit should be ingested.
+	Queue string
+}
+
+// IngestResult is the outcome of a successful ingest operation.
+type IngestResult struct {
+	// ID is the globally unique request identifier assigned to the ingested commit.
+	// Format: "request/<queue>/<counter_value>".
+	ID string
+}