feat(hook): Deliver events to integrations

Author: prathshenoy

Makefile

@@ -377,7 +377,7 @@ local-stovepipe-stop: ## Stop the Stovepipe service
 
 mocks: ## Generate mock files using mockgen
 	@echo "Generating mocks..."
-	@$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/consumergate/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./submitqueue/extension/mergechecker/... ./submitqueue/extension/scorer/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/... ./submitqueue/extension/validator/... ./platform/consumer/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/...
+	@$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/consumergate/... ./platform/extension/hook/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./submitqueue/extension/mergechecker/... ./submitqueue/extension/scorer/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/... ./submitqueue/extension/validator/... ./platform/consumer/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/...
 	@echo "Mocks generated successfully!"
 
 proto: ## Generate protobuf files from .proto definitions

api/base/hook/README.md

@@ -40,9 +40,9 @@ The binding between a topic key and its payload lives in the message's `topic_ke
 
 | Message | Direction | Topic key |
 |---|---|---|
-| `HookEvent` | producing domain → hook dispatcher | `hook` |
+| `HookEvent` | producing domain → hook stage | `hook` |
 
-The key is per-host: each domain runs its own hook topic and its own dispatcher, so two domains sharing one queue backend must map `hook` to distinct topic names.
+The key is per-host: each domain runs its own hook topic and its own hook controller, so two domains sharing one queue backend must map `hook` to distinct topic names.
 
 ## Evolution
 

doc/rfc/hook-framework.md

@@ -10,13 +10,13 @@ Two requirements: side effects must never stall or fail the pipeline, and "fire
 
 ## Proposal
 
-When a controller performs a transition, it also publishes a **hook event** to a durable `hook` topic. A thin per-domain dispatcher stage consumes it and hands each event to the **hooks** the host wired — no-op by default, real integrations as they arrive.
+When a controller performs a transition, it also publishes a **hook event** to a durable `hook` topic. A thin per-domain hook stage consumes it, asks the host's **hooks resolver** which integrations that event belongs to, and runs them — none by default, real integrations as they arrive.
 
 ```
-pipeline controller                              dispatcher stage (per domain)
-state write → hook publish → downstream ──▶ [hook topic] ──▶ decode → validate → hook.Handle
-                                                 │                         ├─ noop (default)
-                                                 │ retries exhausted       └─ composite ─▶ warehouse, code host, …
+pipeline controller                              hook stage (per domain)
+state write → hook publish → downstream ──▶ [hook topic] ──▶ decode → validate → Hooks.For(event)
+                                                 │                                   └─▶ warehouse, code host, …
+                                                 │ retries exhausted
                                                  ▼
                                              [hook_dlq] ──▶ log full event + page; manual republish
 ```
@@ -58,12 +58,13 @@ Delivery promise:
 
 ### Hooks and dispatch
 
-- Extension at `platform/extension/hook/`, singleton shape (counter precedent), wired once per host; no per-queue factory.
+- Extension at `platform/extension/hook/`: the `Hook` contract plus a `Hooks` resolver the host builds in wiring. No `Config` and no `Factory` — selection is the resolver's, and only wiring knows the queue topology.
 - Hook contract: at-least-once, idempotent by `id`, plain errors, never writes pipeline state; ignore an event by returning nil (no filter API).
-- Ships `noop` (default) and `composite` (runs all children, joins failures, names failing children). A cross-domain sink is the same impl wired into each domain.
-- Dispatcher: decode, validate (`id`/`source`/`type` non-empty), invoke. Malformed events dead-letter, never silently acked; hook errors retry then dead-letter, with errs classifiers fast-pathing permanent failures.
+- `Hooks.For(event)` keys on the event, not a queue name: the envelope carries no queue, and which scope selects hooks (queue, source, type) differs per domain. Resolving to none is ordinary. Ships `noop` for a host that wants an explicit placeholder. A cross-domain sink is the same impl wired into each domain.
+- Controller (`platform/hook`, wired by each service): decode, validate (`id`/`source`/`type` non-empty), resolve, invoke all. Malformed events dead-letter, never silently acked; hook errors retry then dead-letter, with errs classifiers fast-pathing permanent failures.
+- Mixed outcomes: every resolved hook runs even after one fails, and the failures are attributed and joined. `errs` weighs each branch of a joined error, so a transient failure alongside a permanent one still retries.
 - DLQ reconciler: log the full event with its failure attribution, page (new metric — the log DLQ only warns), then ack. Manual republish recovers; pipeline state is never touched.
-- Per-hook retry isolation later: consumer groups on the same `hook` topic key, once the registry supports multiple groups per key and rejection becomes group-local (today it moves the shared row). Until then the composite's shared budget is accepted.
+- Per-hook retry isolation later: consumer groups on the same `hook` topic key, once the registry supports multiple groups per key and rejection becomes group-local (today it moves the shared row). Until then one shared budget for all of an event's hooks is accepted.
 
 ## Example
 

platform/extension/hook/BUILD.bazel

@@ -0,0 +1,9 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["hook.go"],
+    importpath = "github.com/uber/submitqueue/platform/extension/hook",
+    visibility = ["//visibility:public"],
+    deps = ["//api/base/hook:go_default_library"],
+)

platform/extension/hook/README.md

@@ -0,0 +1,43 @@
+# Hook
+
+Vendor-agnostic interface for fire-and-forget side effects run in response to pipeline lifecycle events: warehouse exports, code-host comments, notifications, audit trails. See [the hooks framework RFC](../../../doc/rfc/hook-framework.md) for the design and [`api/base/hook`](../../../api/base/hook) for the event contract.
+
+## Interface
+
+### Hook
+
+Handles one lifecycle event. `Name` identifies it in logs, metrics, and failure attribution.
+
+Four obligations, all of them consequences of running behind an at-least-once queue:
+
+- **Idempotent on the event id.** The same event may arrive more than once, including after a successful `Handle`. The id is derived from the transition, so a redelivery carries the id the first delivery did.
+- **Return nil to ignore an event.** There is no filter or subscription API. A hook that does not care about a type returns nil and costs nothing; routing can become a wiring decorator if it ever pays for itself.
+- **Return plain errors.** Classification is the consumer's job. An error must mean the side effect did not happen — reporting failure for work that succeeded turns at-least-once delivery into repeated duplicate effects.
+- **Never write pipeline state.** A hook's outcome is invisible to the pipeline, which is exactly what makes it unable to affect the transition that triggered it.
+
+### Hooks
+
+Resolves the hooks that run for an event. The controller in [`platform/hook`](../../hook) asks it once per delivery and runs everything it returns; returning none is ordinary and means nothing this host wired cares about the event.
+
+`For` takes the event rather than a queue name because the envelope carries no queue. Which scope selects hooks differs per domain — queue, source, event type — and only the host that publishes the payload can read a queue out of it, so the choice belongs to the resolver. Resolution runs on every delivery and cannot fail: an integration that cannot be reached is a `Handle` error, not an absent hook.
+
+## Wiring
+
+There is no `Config` and no `Factory` here. Selection is the resolver's job, and the resolver is built in the wiring layer — the only place that knows the full set of queues and the integrations wired for each. The host constructs its `Hooks` and hands it to the controller in [`platform/hook`](../../hook), which owns the consumer side: decode, validate, resolve, invoke.
+
+Two queues in one host can point at different providers and want different integrations, which is why hooks are resolved per event rather than fixed per deployment.
+
+## Implementations
+
+- **`noop/`** — accepts every event and does nothing. A placeholder for a host that wants the stage registered before it has any integration; a resolver that returns no hooks does the same thing.
+
+A sink that serves several domains is one implementation wired into each domain's host, not one implementation per domain.
+
+## Implementing a Hook
+
+1. Create `platform/extension/hook/{name}/` for a hook reusable across domains, or `{domain}/extension/hook/{name}/` for one that is domain-specific.
+2. Implement `Handle` and `Name`, keying any deduplication on `event.GetId()`.
+3. Decide per event `type` what to do, and return nil for the types you ignore.
+4. Return it from the host's `Hooks` resolver for the events it should run on.
+
+Every hook the resolver returns for an event shares one consumer and therefore one retry budget: one chronically failing integration eventually dead-letters events the others handled fine. See [`platform/hook`](../../hook) before wiring several.

platform/extension/hook/hook.go

@@ -0,0 +1,80 @@
+// 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 hook defines the contract for a hook: a pluggable side effect run in
+// response to a pipeline lifecycle event. Warehouse exports, code-host comments,
+// notifications, and audit trails are all hooks.
+//
+// Which hooks run is a property of the event rather than of the deployment: two
+// queues in one host can point at different providers and want different
+// integrations. A host therefore supplies a Hooks resolver, and the controller
+// in platform/hook asks it once per event.
+//
+// Hooks run behind a durable queue, never inline in the pipeline, so a slow or
+// failing integration cannot stall or fail the work that triggered it.
+package hook
+
+//go:generate mockgen -source=hook.go -destination=mock/hook_mock.go -package=mock
+
+import (
+	"context"
+
+	basehook "github.com/uber/submitqueue/api/base/hook"
+)
+
+// Hook performs a side effect in response to a lifecycle event.
+type Hook interface {
+	// Handle performs the side effect for event.
+	//
+	// Delivery is at-least-once, so the same event — identical id — may arrive
+	// more than once, including after a successful Handle. Implementations must
+	// be idempotent on the event id.
+	//
+	// Returning nil means "done with this event", which is also how a hook
+	// ignores one: there is no filter or subscription API, because a hook that
+	// does not care about a type simply returns nil, and routing can be added as
+	// a wiring decorator if it ever pays for itself.
+	//
+	// Returning an error retries the event and, past the retry budget,
+	// dead-letters it. Return plain errors; classification is the consumer's
+	// job. An error must mean the side effect did not happen — reporting failure
+	// for work that succeeded turns at-least-once into repeated duplicate
+	// effects.
+	//
+	// A hook must never write pipeline state. Its outcome is invisible to the
+	// pipeline by design: that is what makes the side effect unable to affect
+	// the transition that triggered it.
+	Handle(ctx context.Context, event *basehook.HookEvent) error
+
+	// Name identifies the hook in logs, metrics, and the failure attribution
+	// the controller reports. Stable and unique among the hooks a host wires.
+	Name() string
+}
+
+// Hooks resolves the hooks that run for an event.
+type Hooks interface {
+	// For returns the hooks to run for event, in the order they should run.
+	// Returning none is an ordinary outcome: it means nothing this host wired
+	// is interested in the event.
+	//
+	// It takes the event rather than a queue name because the envelope carries
+	// no queue. Which scope selects hooks differs per domain — queue, source,
+	// event type — and only the host that publishes the payload can read a
+	// queue out of it, so the choice belongs to the resolver.
+	//
+	// Called on every delivery, so resolution must be cheap and must not fail:
+	// an integration that cannot be reached is a Handle error, not an absent
+	// hook.
+	For(event *basehook.HookEvent) []Hook
+}

platform/extension/hook/mock/BUILD.bazel

@@ -0,0 +1,13 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["hook_mock.go"],
+    importpath = "github.com/uber/submitqueue/platform/extension/hook/mock",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//api/base/hook:go_default_library",
+        "//platform/extension/hook:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
+    ],
+)

platform/extension/hook/mock/hook_mock.go

@@ -0,0 +1,22 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["hook.go"],
+    importpath = "github.com/uber/submitqueue/platform/extension/hook/noop",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//api/base/hook:go_default_library",
+        "//platform/extension/hook:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["hook_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//api/base/hook:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+    ],
+)

platform/extension/hook/noop/BUILD.bazel

@@ -0,0 +1,44 @@
+// 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 noop provides a hook.Hook that accepts every event and does nothing.
+// It is a placeholder for a host that wants the stage registered before it has
+// any integration — a resolver returning no hooks does the same thing. Either
+// way events are still published, consumed, and acked, so turning a real hook on
+// later changes only what happens to the event, not whether the seam works.
+package noop
+
+import (
+	"context"
+
+	basehook "github.com/uber/submitqueue/api/base/hook"
+	"github.com/uber/submitqueue/platform/extension/hook"
+)
+
+// Verify interface compliance at compile time.
+var _ hook.Hook = Hook{}
+
+// Hook is a hook that discards every event.
+type Hook struct{}
+
+// New returns a no-op Hook.
+func New() Hook {
+	return Hook{}
+}
+
+// Handle implements hook.Hook. The event is discarded.
+func (Hook) Handle(context.Context, *basehook.HookEvent) error { return nil }
+
+// Name implements hook.Hook.
+func (Hook) Name() string { return "noop" }