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

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,36 @@
+# 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.
+
+## Wiring
+
+A hook is wired **once per host**, not resolved per queue, so this package has no `Config` and no `Factory`. What an integration does is a property of the deployment rather than of the queue an event came from; a hook that genuinely needs per-queue behavior resolves the queue from the event payload.
+
+The host constructs its hook and hands it to the dispatcher in [`platform/hook`](../../hook), which owns the consumer side: decode, validate, invoke.
+
+## Implementations
+
+- **`noop/`** — accepts every event and does nothing. The default before a host has any integration, so the seam behaves identically whether or not hooks are configured.
+- **`composite/`** — fans an event out to several children, runs all of them even after one fails, and joins the failures with the name of each failing child. Read its package doc before wiring more than one child: they share a single retry budget, so one chronically failing integration eventually dead-letters events the others handled fine.
+
+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. Wire it into the host's dispatcher — inside a `composite` if the host has more than one.

platform/extension/hook/composite/BUILD.bazel

@@ -0,0 +1,24 @@
+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/composite",
+    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",
+        "//platform/extension/hook:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+    ],
+)

platform/extension/hook/composite/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 composite provides a hook.Hook that fans one event out to several
+// children. It is how a host wires more than one integration, since the
+// dispatcher takes a single hook.
+//
+// Every child runs on every event, even after one fails, so a broken
+// integration cannot stop the others from seeing the event. Failures are
+// collected and joined, each wrapped with the name of the child that raised it,
+// so the error reaching the dispatcher says which integration failed rather than
+// just that something did.
+//
+// # Children share one retry budget
+//
+// The composite is a single consumer, so a retry re-delivers the event to every
+// child, including the ones that already succeeded. Two consequences: children
+// must be idempotent on the event id (the hook contract requires this anyway),
+// and one persistently failing child spends the budget for all of them, so the
+// event eventually dead-letters even though the others were fine.
+//
+// The fix is a consumer group per hook on the shared hook topic, which the queue
+// cannot express today: the registry admits one consumer group per topic key,
+// and a rejection moves the shared message row to the DLQ for every group rather
+// than only the one that rejected it. Until both change, prefer wiring children
+// whose failure modes are independent and short-lived, and treat a chronically
+// failing integration as something to remove from the composite rather than to
+// absorb.
+package composite
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	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 fans an event out to every child hook.
+type Hook struct {
+	// children are the hooks the event is handed to, in wiring order.
+	children []hook.Hook
+}
+
+// New returns a Hook that hands each event to every child in the order given.
+// With no children it accepts every event and does nothing.
+func New(children ...hook.Hook) Hook {
+	return Hook{children: children}
+}
+
+// Handle implements hook.Hook. It runs every child and returns the joined
+// failures, each attributed to the child that raised it, or nil when all
+// succeeded.
+func (h Hook) Handle(ctx context.Context, event *basehook.HookEvent) error {
+	var failures []error
+	for _, child := range h.children {
+		if err := child.Handle(ctx, event); err != nil {
+			failures = append(failures, fmt.Errorf("hook %s: %w", child.Name(), err))
+		}
+	}
+	return errors.Join(failures...)
+}
+
+// Name implements hook.Hook.
+func (Hook) Name() string { return "composite" }

platform/extension/hook/composite/hook_test.go

@@ -0,0 +1,87 @@
+// 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 composite
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	basehook "github.com/uber/submitqueue/api/base/hook"
+	"github.com/uber/submitqueue/platform/extension/hook"
+)
+
+// recordingHook records the events it saw and fails with a fixed error.
+type recordingHook struct {
+	name string
+	err  error
+	seen []string
+}
+
+var _ hook.Hook = (*recordingHook)(nil)
+
+func (h *recordingHook) Handle(_ context.Context, event *basehook.HookEvent) error {
+	h.seen = append(h.seen, event.GetId())
+	return h.err
+}
+
+func (h *recordingHook) Name() string { return h.name }
+
+func event() *basehook.HookEvent {
+	return &basehook.HookEvent{Id: "submitqueue/batch.failed/batch-778/4", Source: "submitqueue", Type: "batch.failed"}
+}
+
+func TestHandle(t *testing.T) {
+	t.Run("no children", func(t *testing.T) {
+		require.NoError(t, New().Handle(context.Background(), event()))
+	})
+
+	t.Run("every child sees the event", func(t *testing.T) {
+		first := &recordingHook{name: "first"}
+		second := &recordingHook{name: "second"}
+
+		require.NoError(t, New(first, second).Handle(context.Background(), event()))
+		assert.Equal(t, []string{event().GetId()}, first.seen)
+		assert.Equal(t, []string{event().GetId()}, second.seen)
+	})
+
+	t.Run("a failing child does not stop the others", func(t *testing.T) {
+		boom := errors.New("boom")
+		failing := &recordingHook{name: "failing", err: boom}
+		healthy := &recordingHook{name: "healthy"}
+
+		err := New(failing, healthy).Handle(context.Background(), event())
+
+		require.Error(t, err)
+		assert.ErrorIs(t, err, boom)
+		assert.Equal(t, []string{event().GetId()}, healthy.seen, "the healthy child runs after the failing one")
+	})
+
+	t.Run("every failure survives the join", func(t *testing.T) {
+		first := errors.New("first failure")
+		second := errors.New("second failure")
+
+		err := New(
+			&recordingHook{name: "first", err: first},
+			&recordingHook{name: "second", err: second},
+		).Handle(context.Background(), event())
+
+		require.Error(t, err)
+		assert.ErrorIs(t, err, first)
+		assert.ErrorIs(t, err, second)
+	})
+}

platform/extension/hook/hook.go

@@ -0,0 +1,65 @@
+// 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.
+//
+// A hook is wired once per host rather than resolved per queue, because what an
+// integration does — post a comment, write a row — is a property of the
+// deployment, not of the queue the event came from. There is therefore no Config
+// and no Factory here: the host constructs its hook directly and hands it to the
+// dispatcher. A hook that genuinely needs per-queue behavior resolves the queue
+// from the event payload.
+//
+// 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 a
+	// composite reports. Stable and unique among the hooks a host wires.
+	Name() string
+}

platform/extension/hook/mock/BUILD.bazel

@@ -0,0 +1,12 @@
+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",
+        "@org_uber_go_mock//gomock:go_default_library",
+    ],
+)