feat(stovepipe): Consume hook events (#632)

## Summary
**What**:
- Register the hook stage and its dead-letter stage in Stovepipe's
consumer wiring, on a topic named for the domain, with a resolver that
hands every event to the no-op hook.

**Why**:
- Wire the stage in a second domain, so attaching a real integration to
Stovepipe is a resolver swap rather than new plumbing, and two domains
on one queue backend keep their own hook topics.

## Test Plan
- [x] Add unit tests.

## Revert Plan
- Revert this PR. Nothing publishes hook events yet and the resolver
returns only a no-op hook.

## Issues
-
[CODEM-416](https://linear.app/uber/issue/CODEM-416/hooks-integration-downstream-notificaiton)

## Stack
1. #607
1. #608
1. @ #632

Author: prathshenoy

service/stovepipe/README.md

@@ -7,7 +7,8 @@ Runnable wiring for the **Stovepipe** domain — a single-service domain (the do
 - **process consumer** (`TopicKeyProcess`) — reloads the persisted `Request` from storage and runs the process stage (`stovepipe/controller/process`).
 - **build consumer** (`TopicKeyBuild`) — reloads the persisted `Request` and triggers the build-runner, then publishes to `buildsignal`.
 - **buildsignal consumer** (`TopicKeyBuildSignal`) — polls/records the build's terminal status and releases the queue's in-flight slot, then publishes to `record`.
-- **record consumer** (`TopicKeyRecord`) — writes the whole-repo validation fact, advances the queue's last-green bookmark and promotion ref, and publishes hook events.
+- **record consumer** (`TopicKeyRecord`) — writes the whole-repo validation fact, and advances the queue's last-green bookmark and promotion ref.
+- **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.
 - **DLQ reconciler** — for each internal topic, a `_dlq` consumer that drives stuck requests to a conservative terminal state so the queue's slot is freed.
 
 The ingest → process → build → buildsignal → record hop stays inside one service and one store, so the queue messages carry only request **IDs**; the consumers reload from storage (the source of truth), which keeps messages small and redelivery idempotent. The process, build, buildsignal, and record topic keys and their internal wire contract are owned by the domain under `stovepipe/core/messagequeue/`.
@@ -43,7 +44,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)
+		}
+	})
+}