feat(stovepipe): Consume hook events

**What**:
- Consume lifecycle events on Stovepipe's own hook topic and hand each one to the integrations it resolves to, accepting and discarding them while only the no-op integration is attached.
- Record and alert on an event that fails every retry, rather than letting it accumulate unread.

**Why**:
- Show the delivery stage works for a second domain, one that registers its consumers directly rather than through the pipeline topology.
- Keep two domains sharing a queue backend from consuming each other's events.

Author: prathshenoy

service/stovepipe/README.md

@@ -1,10 +1,11 @@
 # Stovepipe Service
 
-Runnable wiring for the **Stovepipe** domain — a single-service domain (the domain *is* the service). The server exposes two RPCs and runs one internal pipeline stage as a queue consumer:
+Runnable wiring for the **Stovepipe** domain — a single-service domain (the domain *is* the service). The server exposes two RPCs and runs its internal pipeline stages as queue consumers:
 
 - **`Ping`** — health check.
 - **`Ingest`** — resolves a queue's head commit, persists a `Request` (and its head URI) to storage, and publishes the request to the **process** stage.
 - **process consumer** (`TopicKeyProcess`) — reloads the persisted `Request` from storage and runs the process stage (`stovepipe/controller/process`).
+- **hook consumer** (`TopicKeyHook`) — hands each lifecycle event to the hooks `hookResolver` returns (`platform/hook`). Nothing publishes to this topic yet and the resolver returns only `noop`, so events are accepted and discarded. Its topic name is domain-qualified (`stovepipe-hook`) because the key is shared across domains.
 
 The ingest → process hop stays inside one service and one store, so only the request **ID** travels on the queue; the consumer reloads from storage (the source of truth), which keeps messages small and redelivery idempotent. The process topic key and its internal wire contract are owned by the domain under `stovepipe/core/messagequeue/`.
 
@@ -23,7 +24,7 @@ Stovepipe therefore needs two MySQL databases: a **storage** database (the `requ
 stovepipe/
 ├── docker-compose.yml      # Stovepipe service + storage MySQL + queue MySQL
 ├── server/
-│   ├── main.go             # gRPC server (Ping, Ingest) + process-stage consumer wiring
+│   ├── main.go             # gRPC server (Ping, Ingest) + queue consumer wiring
 │   └── Dockerfile
 └── client/
     └── main.go             # Ping client (default :8083)
@@ -38,7 +39,7 @@ The Stovepipe controllers live under [`stovepipe/controller/`](../../stovepipe/c
 | `STORAGE_MYSQL_DSN` | yes      | Storage database DSN (`request`, `request_uri`) | —             |
 | `QUEUE_MYSQL_DSN`   | yes      | Queue database DSN                       | —                    |
 | `PORT`              | no       | gRPC listen address                      | `:8083`              |
-| `HOSTNAME`          | no       | Subscriber name for the process consumer | `stovepipe-<unix_ts>` |
+| `HOSTNAME`          | no       | Subscriber name for the queue consumers  | `stovepipe-<unix_ts>` |
 
 ## Running
 

service/stovepipe/server/BUILD.bazel

@@ -1,11 +1,12 @@
-load("@rules_go//go:def.bzl", "go_binary", "go_cross_binary", "go_library")
+load("@rules_go//go:def.bzl", "go_binary", "go_cross_binary", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
     srcs = ["main.go"],
     importpath = "github.com/uber/submitqueue/service/stovepipe/server",
     visibility = ["//visibility:private"],
     deps = [
+        "//api/base/hook:go_default_library",
         "//api/stovepipe/protopb:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/errs:go_default_library",
@@ -14,8 +15,11 @@ go_library(
         "//platform/errs/mysql:go_default_library",
         "//platform/extension/consumergate/noop:go_default_library",
         "//platform/extension/counter:go_default_library",
+        "//platform/extension/hook:go_default_library",
+        "//platform/extension/hook/noop:go_default_library",
         "//platform/extension/messagequeue:go_default_library",
         "//platform/extension/messagequeue/mysql:go_default_library",
+        "//platform/hook:go_default_library",
         "//service/stovepipe/server/mapper:go_default_library",
         "//stovepipe/controller:go_default_library",
         "//stovepipe/controller/build:go_default_library",
@@ -66,3 +70,18 @@ filegroup(
     ],
     visibility = ["//test:__subpackages__"],
 )
+
+go_test(
+    name = "go_default_test",
+    srcs = ["main_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//api/base/hook:go_default_library",
+        "//platform/consumer:go_default_library",
+        "//stovepipe/controller/dlq: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_zap//zaptest:go_default_library",
+    ],
+)

service/stovepipe/server/main.go

@@ -28,6 +28,7 @@ import (
 
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/uber-go/tally"
+	basehook "github.com/uber/submitqueue/api/base/hook"
 	pb "github.com/uber/submitqueue/api/stovepipe/protopb"
 	"github.com/uber/submitqueue/platform/consumer"
 	"github.com/uber/submitqueue/platform/errs"
@@ -36,8 +37,11 @@ import (
 	mysqlerrs "github.com/uber/submitqueue/platform/errs/mysql"
 	consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop"
 	"github.com/uber/submitqueue/platform/extension/counter"
+	hookext "github.com/uber/submitqueue/platform/extension/hook"
+	hooknoop "github.com/uber/submitqueue/platform/extension/hook/noop"
 	extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
 	queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql"
+	platformhook "github.com/uber/submitqueue/platform/hook"
 	"github.com/uber/submitqueue/service/stovepipe/server/mapper"
 	"github.com/uber/submitqueue/stovepipe/controller"
 	"github.com/uber/submitqueue/stovepipe/controller/build"
@@ -149,6 +153,16 @@ func (fakeBuildRunnerFactory) For(cfg buildrunner.Config) (buildrunner.BuildRunn
 	return buildrunnerfake.New(cfg), nil
 }
 
+// hookResolver sends every event to the no-op hook. Which hooks an event goes to is host
+// policy, so the resolver lives here rather than in the extension package. A deployment
+// with real integrations swaps this for one that selects on the event's source and type.
+type hookResolver struct{}
+
+// For returns the hooks that run for event.
+func (hookResolver) For(*basehook.HookEvent) []hookext.Hook {
+	return []hookext.Hook{hooknoop.New()}
+}
+
 func main() {
 	code := 0
 	if err := run(); err != nil {
@@ -286,7 +300,7 @@ func run() error {
 	brf := fakeBuildRunnerFactory{}
 
 	storageFty := storageFactory{backend: store}
-	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, registry, sourceControl, brf)
+	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, registry, sourceControl, brf, hookResolver{})
 	if err != nil {
 		return err
 	}
@@ -400,6 +414,7 @@ func registerPrimaryControllers(
 	registry consumer.TopicRegistry,
 	sourceControl sourcecontrol.Factory,
 	brf buildrunner.Factory,
+	hooks hookext.Hooks,
 ) (int, error) {
 	var count int
 
@@ -436,6 +451,12 @@ func registerPrimaryControllers(
 	}
 	count++
 
+	hookController := platformhook.NewController(logger, scope, hooks, basehook.TopicKeyHook, "stovepipe-hook")
+	if err := c.Register(hookController); err != nil {
+		return count, fmt.Errorf("failed to register hook controller: %w", err)
+	}
+	count++
+
 	return count, nil
 }
 
@@ -475,6 +496,12 @@ func registerDLQControllers(
 	}
 	count++
 
+	hookDLQController := platformhook.NewDLQController(logger, scope, dlq.TopicKey(basehook.TopicKeyHook), "stovepipe-hook-dlq")
+	if err := c.Register(hookDLQController); err != nil {
+		return count, fmt.Errorf("failed to register hook dlq controller: %w", err)
+	}
+	count++
+
 	return count, nil
 }
 
@@ -484,6 +511,9 @@ func registerDLQControllers(
 // topic and the buildsignal consumer subscribes to it, and also republishes to itself while
 // polling. buildsignal publishes to the record topic once a build reaches a terminal status,
 // and the record consumer subscribes to it.
+//
+// The hook topic name is domain-qualified because its key is shared across domains: two
+// domains pointed at one queue backend would otherwise consume each other's events.
 func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) {
 	return consumer.NewTopicRegistry([]consumer.TopicConfig{
 		{
@@ -518,6 +548,14 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 				subscriberName, "stovepipe-record",
 			),
 		},
+		{
+			Key:   basehook.TopicKeyHook,
+			Name:  "stovepipe-hook",
+			Queue: q,
+			Subscription: extqueue.DefaultSubscriptionConfig(
+				subscriberName, "stovepipe-hook",
+			),
+		},
 		{
 			Key:          dlq.TopicKey(stovepipemq.TopicKeyProcess),
 			Name:         "process_dlq",
@@ -542,6 +580,12 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 			Queue:        q,
 			Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-record-dlq"),
 		},
+		{
+			Key:          dlq.TopicKey(basehook.TopicKeyHook),
+			Name:         "stovepipe-hook_dlq",
+			Queue:        q,
+			Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-hook-dlq"),
+		},
 	})
 }
 

service/stovepipe/server/main_test.go

@@ -0,0 +1,112 @@
+// Copyright (c) 2026 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 main
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/uber-go/tally"
+	basehook "github.com/uber/submitqueue/api/base/hook"
+	"github.com/uber/submitqueue/platform/consumer"
+	"github.com/uber/submitqueue/stovepipe/controller/dlq"
+	"go.uber.org/zap/zaptest"
+)
+
+// recordingConsumer captures what the host registers instead of subscribing.
+type recordingConsumer struct {
+	controllers []consumer.Controller
+}
+
+func (c *recordingConsumer) Register(controller consumer.Controller) error {
+	c.controllers = append(c.controllers, controller)
+	return nil
+}
+
+func (c *recordingConsumer) Start(context.Context) error { return nil }
+
+func (c *recordingConsumer) Stop(int64) error { return nil }
+
+// registeredControllers runs the host's registration exactly as run() does and
+// returns the registry it registers against, the primary controllers, and the
+// DLQ controllers.
+func registeredControllers(t *testing.T) (consumer.TopicRegistry, []consumer.Controller, []consumer.Controller) {
+	t.Helper()
+
+	registry, err := newTopicRegistry(nil, "subscriber")
+	require.NoError(t, err)
+
+	logger := zaptest.NewLogger(t).Sugar()
+	store := storageFactory{}
+	primary := &recordingConsumer{}
+	deadLetter := &recordingConsumer{}
+
+	_, err = registerPrimaryControllers(primary, logger, tally.NoopScope, store, registry,
+		fakeSourceControlFactory{}, fakeBuildRunnerFactory{}, hookResolver{})
+	require.NoError(t, err)
+
+	_, err = registerDLQControllers(deadLetter, logger, tally.NoopScope, store, registry,
+		fakeSourceControlFactory{})
+	require.NoError(t, err)
+
+	return registry, primary.controllers, deadLetter.controllers
+}
+
+func topicKeys(controllers []consumer.Controller) []consumer.TopicKey {
+	keys := make([]consumer.TopicKey, 0, len(controllers))
+	for _, c := range controllers {
+		keys = append(keys, c.TopicKey())
+	}
+	return keys
+}
+
+func TestEveryRegisteredControllerResolvesInTheTopicRegistry(t *testing.T) {
+	registry, primary, deadLetter := registeredControllers(t)
+
+	for _, c := range append(primary, deadLetter...) {
+		t.Run(c.Name(), func(t *testing.T) {
+			_, ok := registry.TopicName(c.TopicKey())
+			assert.True(t, ok, "no topic name registered for the key the controller subscribes to")
+
+			_, ok = registry.SubscriptionConfig(c.TopicKey(), c.ConsumerGroup())
+			assert.True(t, ok, "no subscription registered for the controller's consumer group")
+		})
+	}
+}
+
+func TestHookStage(t *testing.T) {
+	registry, primary, deadLetter := registeredControllers(t)
+
+	t.Run("hook events are consumed", func(t *testing.T) {
+		assert.Contains(t, topicKeys(primary), basehook.TopicKeyHook)
+	})
+
+	t.Run("hook events that exhaust their retries are consumed", func(t *testing.T) {
+		assert.Contains(t, topicKeys(deadLetter), dlq.TopicKey(basehook.TopicKeyHook))
+	})
+
+	// The key is shared across domains, so an unqualified name would collide with
+	// another domain's hook topic on a queue backend the two share.
+	t.Run("the hook topics are named for this domain", func(t *testing.T) {
+		for _, key := range []consumer.TopicKey{basehook.TopicKeyHook, dlq.TopicKey(basehook.TopicKeyHook)} {
+			name, ok := registry.TopicName(key)
+			require.True(t, ok)
+			assert.True(t, strings.HasPrefix(name, "stovepipe-"), "topic %q is not domain-qualified", name)
+		}
+	})
+}