From b5394ebea7d0d0b923994bd96ed8d4e63e32c381 Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Mon, 17 Aug 2026 14:25:22 +0000 Subject: [PATCH] feat(hook): Add contract for lifecycle events --- Makefile | 2 +- api/base/hook/BUILD.bazel | 31 +++++ api/base/hook/README.md | 49 +++++++ api/base/hook/event.go | 72 ++++++++++ api/base/hook/hook.go | 62 +++++++++ api/base/hook/hook_test.go | 184 ++++++++++++++++++++++++++ api/base/hook/proto/BUILD.bazel | 4 + api/base/hook/proto/hook.proto | 58 +++++++++ api/base/hook/protopb/BUILD.bazel | 14 ++ api/base/hook/protopb/hook.pb.go | 210 ++++++++++++++++++++++++++++++ api/base/hook/topics.go | 26 ++++ tool/proto/BUILD.bazel | 11 ++ 12 files changed, 722 insertions(+), 1 deletion(-) create mode 100644 api/base/hook/BUILD.bazel create mode 100644 api/base/hook/README.md create mode 100644 api/base/hook/event.go create mode 100644 api/base/hook/hook.go create mode 100644 api/base/hook/hook_test.go create mode 100644 api/base/hook/proto/BUILD.bazel create mode 100644 api/base/hook/proto/hook.proto create mode 100644 api/base/hook/protopb/BUILD.bazel create mode 100644 api/base/hook/protopb/hook.pb.go create mode 100644 api/base/hook/topics.go diff --git a/Makefile b/Makefile index ff8fe283..e393ba87 100644 --- a/Makefile +++ b/Makefile @@ -36,7 +36,7 @@ GOIMPORTS_VERSION ?= v0.33.0 # (the out_dir convention in tool/proto/BUILD.bazel) and copied back here. A # package may hold multiple .proto files (e.g. an RPC contract plus messagequeue # contracts); all generated stubs land in the same protopb/ dir. -PROTO_PACKAGES = api/base/change api/base/mergestrategy api/base/messagequeue api/runway/messagequeue api/runway api/submitqueue/gateway api/submitqueue/orchestrator api/stovepipe stovepipe/core/messagequeue +PROTO_PACKAGES = api/base/change api/base/hook api/base/mergestrategy api/base/messagequeue api/runway/messagequeue api/runway api/submitqueue/gateway api/submitqueue/orchestrator api/stovepipe stovepipe/core/messagequeue # Set REPO_ROOT for docker-compose export REPO_ROOT := $(shell pwd) diff --git a/api/base/hook/BUILD.bazel b/api/base/hook/BUILD.bazel new file mode 100644 index 00000000..6b15b087 --- /dev/null +++ b/api/base/hook/BUILD.bazel @@ -0,0 +1,31 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = [ + "event.go", + "hook.go", + "topics.go", + ], + importpath = "github.com/uber/submitqueue/api/base/hook", + visibility = ["//visibility:public"], + deps = [ + "//api/base/hook/protopb:go_default_library", + "//api/base/messagequeue/protopb:go_default_library", + "//platform/consumer:go_default_library", + "@org_golang_google_protobuf//encoding/protojson:go_default_library", + "@org_golang_google_protobuf//proto:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["hook_test.go"], + embed = [":go_default_library"], + deps = [ + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@org_golang_google_protobuf//proto:go_default_library", + "@org_golang_google_protobuf//types/known/structpb:go_default_library", + ], +) diff --git a/api/base/hook/README.md b/api/base/hook/README.md new file mode 100644 index 00000000..f4831764 --- /dev/null +++ b/api/base/hook/README.md @@ -0,0 +1,49 @@ +# Hook event contract + +The published, language-neutral contract for hook events: fire-and-forget lifecycle notifications that let integrations react to a pipeline transition without being able to stall or fail the pipeline. See [the hooks framework RFC](../../../doc/rfc/hook-framework.md) for the design and [the message queue contract RFC](../../../doc/rfc/messagequeue-contract.md) for the conventions it follows. + +It lives under `api/base/` rather than `api/{domain}/` because no domain owns it. Every domain publishes this same shape to its own hook topic, so a sink consuming several domains reads one schema rather than one per producer. + +Payloads are defined as proto3 messages in [`proto/hook.proto`](proto/hook.proto) and generated into [`protopb/`](protopb); the proto is the authority and a non-Go client compiles against it directly. On the wire, payloads are serialized as protobuf JSON (`protojson`), so the queue keeps storing self-describing JSON. The Go helpers here are generic `protojson` glue — `Marshal(m)` and `Unmarshal[T](b, m)` — plus the two rules that must be identical across producers: how an id is minted and what makes an event well-formed. Field names stay snake_case (`UseProtoNames`) and `int64` fields serialize as JSON strings. + +## The envelope + +`HookEvent` carries `id`, `source`, `type`, `timestamp_ms`, `version`, and `payload`. The envelope holds only what every consumer keys on uniformly; everything specific to what happened lives in the payload. + +`source` and `type` are open strings rather than enums, and `payload` is a `google.protobuf.Struct` rather than a `oneof`. That is the central trade: a producer adds a new event type by publishing it, instead of by changing the wire contract and redeploying every consumer. protojson rejects unknown *enum* values, so an enum here would break existing consumers on every addition. + +Subject, queue, and error are deliberately **not** on the envelope. They are facts about a particular occurrence, so they belong in the payload — no major event platform carries a top-level error either. + +## Identity and idempotency + +`id` is derived from the transition, not random: `source`, `type`, the subject's id, and the subject's post-transition `version`, joined. `NewEventID` mints it. Replaying the delivery that caused the transition therefore mints the *same* id, which is what lets the queue dedupe the redelivery and lets a hook stay idempotent by keying on it. That derivation is why the framework needs no transactional outbox: the publish rides inside the delivery that performed the state write, and a crash before the ack replays both halves safely. + +When a transition is not a versioned write there is no version to distinguish occurrences, so the id of the message that caused it stands in, plus an ordinal when one cause publishes several same-typed events. `NewUnversionedEventID` mints that form. + +Consumers never parse an id. It is a dedupe and idempotency key, not a structured field. + +## Staleness + +`version` is the subject's optimistic-locking version immediately after the transition, and `0` when the transition was not a versioned write. Delivery is at-least-once, so a hook can receive an event describing a transition that has since been superseded; comparing this version against the subject's current version in the store is how it tells the two apart. Timestamps cannot answer that, because the clocks belong to different machines. + +A domain with no versioned entities (Runway holds no durable state of its own) publishes `0` throughout. That is the normal mode for such a producer, not a degenerate case. + +## Payload + +Shaped per `type` by the domain that publishes it, add-only, and documented by that domain. It must carry the subject's id, and it must carry any fact recorded nowhere else — merge step outcomes, build failure detail — because for those the event is the only durable record. + +It must **not** be an entity snapshot. A snapshot is stale the moment it is redelivered, it competes with the store as a source of truth, and it drags a domain's schema into a contract shared by every domain. Hooks resolve entities from their stores. + +## Topic keys + +The binding between a topic key and its payload lives in the message's `topic_keys` option (defined in `api/base/messagequeue`); `TopicKeys` reads it back by reflection. A topic key is a stable logical name, not a concrete wire topic — each implementer maps the key to whatever topic name its broker/queue requires, via `consumer.TopicRegistry` in our Go wiring. + +| Message | Direction | Topic key | +|---|---|---| +| `HookEvent` | producing domain → hook dispatcher | `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. + +## Evolution + +Contract changes are additive-only: add new fields; never remove, rename, repurpose, or retype an existing field, and never reuse a field number. protojson ignores unknown fields on read and omits zero-valued fields on write, so a new optional field is backward-compatible in both directions. New event types and new payload keys are not contract changes at all — that is the point of the open envelope. diff --git a/api/base/hook/event.go b/api/base/hook/event.go new file mode 100644 index 00000000..5d85868d --- /dev/null +++ b/api/base/hook/event.go @@ -0,0 +1,72 @@ +// 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 + +import ( + "fmt" + "strconv" + "strings" +) + +// Consumers never parse an id, so this is a minting convention rather than a +// wire format. All it must guarantee is that two different transitions cannot +// join to the same string. +const idSeparator = "/" + +// NewEventID mints the id of an event describing a versioned state write. +// +// Deriving the id rather than randomizing it is what makes replay safe: the same +// transition mints the same id, so the queue dedupes a redelivery and a hook +// stays idempotent without a publisher-side outbox. version is the subject's +// version immediately after the write, which is what separates two transitions +// of the same subject. +func NewEventID(source, eventType, subjectID string, version int32) string { + return strings.Join([]string{source, eventType, subjectID, strconv.Itoa(int(version))}, idSeparator) +} + +// NewUnversionedEventID mints the id of an event whose transition was not a +// versioned write, so no version distinguishes one occurrence from the next. +// +// The causing message's id stands in for the version, being stable across +// redeliveries for the same reason a version is. ordinal separates several +// same-typed events published for one cause; pass 0 when there is only one. +func NewUnversionedEventID(source, eventType, subjectID, causeID string, ordinal int) string { + return strings.Join( + []string{source, eventType, subjectID, causeID, strconv.Itoa(ordinal)}, + idSeparator, + ) +} + +// Validate reports whether e carries the three envelope fields every consumer +// keys on. The rest cannot be checked generically: version is legitimately 0 for +// an unversioned transition and payload is shaped per type. +// +// Both sides call it — a publisher to catch a malformed event before it reaches +// the queue, a consumer because the producer may not have. +func Validate(e *HookEvent) error { + if e == nil { + return fmt.Errorf("hook event is nil") + } + if e.GetId() == "" { + return fmt.Errorf("hook event has no id") + } + if e.GetSource() == "" { + return fmt.Errorf("hook event %q has no source", e.GetId()) + } + if e.GetType() == "" { + return fmt.Errorf("hook event %q has no type", e.GetId()) + } + return nil +} diff --git a/api/base/hook/hook.go b/api/base/hook/hook.go new file mode 100644 index 00000000..c3802373 --- /dev/null +++ b/api/base/hook/hook.go @@ -0,0 +1,62 @@ +// 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 holds the hook event contract: the wire payload every domain +// publishes to its own hook topic for fire-and-forget lifecycle side effects. +package hook + +import ( + "google.golang.org/protobuf/encoding/protojson" + "google.golang.org/protobuf/proto" + + "github.com/uber/submitqueue/api/base/hook/protopb" + basemqpb "github.com/uber/submitqueue/api/base/messagequeue/protopb" +) + +// HookEvent aliases the generated binding so callers reference the contract +// through this package rather than protopb. +type HookEvent = protopb.HookEvent + +// UseProtoNames keeps JSON field names snake_case, matching the declared +// contract rather than protojson's default lowerCamelCase. +var marshalOpts = protojson.MarshalOptions{UseProtoNames: true} + +// DiscardUnknown makes an additive contract change backward-compatible: a field +// this consumer does not know yet is ignored rather than rejected. +var unmarshalOpts = protojson.UnmarshalOptions{DiscardUnknown: true} + +// Marshal serializes a contract message to protojson bytes for the queue payload. +func Marshal(m proto.Message) ([]byte, error) { + return marshalOpts.Marshal(m) +} + +// Unmarshal deserializes protojson bytes into the contract message m. +func Unmarshal[T proto.Message](b []byte, m T) error { + return unmarshalOpts.Unmarshal(b, m) +} + +// TopicKeys returns the logical topic keys bound to a message via the +// topic_keys proto option, or nil if it declares none. These are not wire topic +// names; a caller maps each key to its backend's topic. +func TopicKeys(m proto.Message) []string { + opts := m.ProtoReflect().Descriptor().Options() + if opts == nil { + return nil + } + keys, ok := proto.GetExtension(opts, basemqpb.E_TopicKeys).([]string) + if !ok { + return nil + } + return keys +} diff --git a/api/base/hook/hook_test.go b/api/base/hook/hook_test.go new file mode 100644 index 00000000..f3a3b58c --- /dev/null +++ b/api/base/hook/hook_test.go @@ -0,0 +1,184 @@ +// 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 + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/structpb" +) + +func mustStruct(t *testing.T, fields map[string]any) *structpb.Struct { + t.Helper() + s, err := structpb.NewStruct(fields) + require.NoError(t, err) + return s +} + +func TestHookEventRoundTrip(t *testing.T) { + cases := map[string]*HookEvent{ + "versioned with nested payload": { + Id: "submitqueue/batch.failed/batch-778/4", + Source: "submitqueue", + Type: "batch.failed", + TimestampMs: 1722800012345, + Version: 4, + Payload: mustStruct(t, map[string]any{ + "batch_id": "batch-778", + "queue": "go-monorepo", + "error": "merge conflict", + "failed_step": "sq-12346", + "conflict_paths": []any{"foo/bar.go"}, + }), + }, + "unversioned": { + Id: "runway/merge.completed/queue-a-42/msg-9/0", + Source: "runway", + Type: "merge.completed", + TimestampMs: 1722800012345, + Payload: mustStruct(t, map[string]any{"request_id": "queue-a/42"}), + }, + "envelope only": { + Id: "stovepipe/commit.green/git-abc/1", + Source: "stovepipe", + Type: "commit.green", + TimestampMs: 1722800012345, + Version: 1, + }, + } + + for name, event := range cases { + t.Run(name, func(t *testing.T) { + data, err := Marshal(event) + require.NoError(t, err) + + got := &HookEvent{} + require.NoError(t, Unmarshal(data, got)) + assert.True(t, proto.Equal(event, got), "round-tripped HookEvent should equal the original") + }) + } +} + +func TestWireFormat(t *testing.T) { + data, err := Marshal(&HookEvent{ + Id: "submitqueue/batch.failed/batch-778/4", + Source: "submitqueue", + Type: "batch.failed", + TimestampMs: 1722800012345, + Version: 4, + }) + require.NoError(t, err) + + assert.Contains(t, string(data), `"timestamp_ms"`, "fields must serialize as snake_case") + assert.Contains(t, string(data), `"1722800012345"`, "int64 must serialize as a JSON string") +} + +func TestUnmarshalDiscardsUnknownFields(t *testing.T) { + data := []byte(`{"id":"a/b/c/1","source":"a","type":"b","field_from_the_future":7}`) + + got := &HookEvent{} + require.NoError(t, Unmarshal(data, got)) + assert.Equal(t, "a/b/c/1", got.GetId()) +} + +func TestTopicKeysBindEveryTopicKey(t *testing.T) { + bound := map[string]int{} + for _, m := range []proto.Message{&HookEvent{}} { + keys := TopicKeys(m) + require.NotEmpty(t, keys, "message must declare a non-empty topic_keys option") + for _, key := range keys { + bound[key]++ + } + } + + keys := []TopicKey{TopicKeyHook} + + valid := map[string]bool{} + for _, k := range keys { + valid[k.String()] = true + assert.Equalf(t, 1, bound[k.String()], "topic key %q must be bound to exactly one message via the topic_keys option", k) + } + for key := range bound { + assert.Truef(t, valid[key], "topic_keys option names unknown key %q", key) + } +} + +func TestEventIDIsDerived(t *testing.T) { + t.Run("same transition mints the same id", func(t *testing.T) { + assert.Equal(t, + NewEventID("submitqueue", "batch.failed", "batch-778", 4), + NewEventID("submitqueue", "batch.failed", "batch-778", 4), + ) + }) + + t.Run("distinct transitions mint distinct ids", func(t *testing.T) { + ids := map[string]string{ + "baseline": NewEventID("submitqueue", "batch.failed", "batch-778", 4), + "later version": NewEventID("submitqueue", "batch.failed", "batch-778", 5), + "other subject": NewEventID("submitqueue", "batch.failed", "batch-779", 4), + "other type": NewEventID("submitqueue", "batch.succeeded", "batch-778", 4), + "other source": NewEventID("stovepipe", "batch.failed", "batch-778", 4), + "unversioned": NewUnversionedEventID("submitqueue", "batch.failed", "batch-778", "msg-1", 0), + "second ordinal": NewUnversionedEventID("submitqueue", "batch.failed", "batch-778", "msg-1", 1), + "other cause": NewUnversionedEventID("submitqueue", "batch.failed", "batch-778", "msg-2", 0), + "slashed subject": NewEventID("stovepipe", "commit.green", "request/monorepo/main/42", 4), + } + + seen := map[string]string{} + for name, id := range ids { + if other, dup := seen[id]; dup { + t.Errorf("%q and %q both mint id %q", name, other, id) + } + seen[id] = name + } + }) +} + +func TestValidate(t *testing.T) { + valid := func() *HookEvent { + return &HookEvent{Id: "submitqueue/batch.failed/batch-778/4", Source: "submitqueue", Type: "batch.failed"} + } + + t.Run("well-formed event", func(t *testing.T) { + require.NoError(t, Validate(valid())) + }) + + t.Run("unversioned event is well-formed", func(t *testing.T) { + event := valid() + event.Version = 0 + require.NoError(t, Validate(event)) + }) + + t.Run("event with no payload is well-formed", func(t *testing.T) { + event := valid() + event.Payload = nil + require.NoError(t, Validate(event)) + }) + + malformed := map[string]*HookEvent{ + "nil": nil, + "no id": {Source: "submitqueue", Type: "batch.failed"}, + "no source": {Id: "submitqueue/batch.failed/batch-778/4", Type: "batch.failed"}, + "no type": {Id: "submitqueue/batch.failed/batch-778/4", Source: "submitqueue"}, + } + for name, event := range malformed { + t.Run(name, func(t *testing.T) { + require.Error(t, Validate(event)) + }) + } +} diff --git a/api/base/hook/proto/BUILD.bazel b/api/base/hook/proto/BUILD.bazel new file mode 100644 index 00000000..189731ca --- /dev/null +++ b/api/base/hook/proto/BUILD.bazel @@ -0,0 +1,4 @@ +exports_files( + ["hook.proto"], + visibility = ["//tool/proto:__pkg__"], +) diff --git a/api/base/hook/proto/hook.proto b/api/base/hook/proto/hook.proto new file mode 100644 index 00000000..26270f23 --- /dev/null +++ b/api/base/hook/proto/hook.proto @@ -0,0 +1,58 @@ +// 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. + +syntax = "proto3"; + +package uber.base.hook; + +import "google/protobuf/struct.proto"; + +import "api/base/messagequeue/proto/messagequeue.proto"; + +option go_package = "github.com/uber/submitqueue/api/base/hook/protopb"; +option java_multiple_files = true; +option java_outer_classname = "HookProto"; +option java_package = "com.uber.submitqueue.base.hook"; + +// HookEvent is one fire-and-forget lifecycle event. Every domain publishes this +// same shape to its own hook topic, so a sink that consumes several domains +// reads one schema rather than one per producer. +message HookEvent { + option (uber.base.messagequeue.topic_keys) = "hook"; + + // id is the opaque identity of this occurrence, derived from the transition + // it describes so that replaying the transition mints the same id. It is + // the queue's dedupe key and a hook's idempotency key, and is never parsed. + string id = 1; + // source is the domain that produced the event. An open string rather than + // an enum so a new producer does not break existing consumers. + string source = 2; + // type is what happened, as one dotted open string: "request.landed", + // "batch.failed", etc. It is the only dimension a consumer filters on, and + // open for the same reason as source. + string type = 3; + // timestamp_ms is when the occurrence happened, in milliseconds since the + // Unix epoch, on the publisher's clock. + int64 timestamp_ms = 4; + // version is the subject's optimistic-locking version immediately after the + // transition, and 0 when the transition was not a versioned write. A hook + // compares it against the subject's current version to tell a current event + // from one that has since been superseded. + int32 version = 5; + // payload carries the facts specific to type, always including the + // subject's id, and must carry any fact recorded nowhere else because the + // event is that fact's only durable record. Add-only, and never an entity + // snapshot: hooks resolve entities from their stores. + google.protobuf.Struct payload = 6; +} diff --git a/api/base/hook/protopb/BUILD.bazel b/api/base/hook/protopb/BUILD.bazel new file mode 100644 index 00000000..574ecdd3 --- /dev/null +++ b/api/base/hook/protopb/BUILD.bazel @@ -0,0 +1,14 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["hook.pb.go"], + importpath = "github.com/uber/submitqueue/api/base/hook/protopb", + visibility = ["//visibility:public"], + deps = [ + "//api/base/messagequeue/protopb:go_default_library", + "@org_golang_google_protobuf//reflect/protoreflect:go_default_library", + "@org_golang_google_protobuf//runtime/protoimpl:go_default_library", + "@org_golang_google_protobuf//types/known/structpb:go_default_library", + ], +) diff --git a/api/base/hook/protopb/hook.pb.go b/api/base/hook/protopb/hook.pb.go new file mode 100644 index 00000000..cc0046c4 --- /dev/null +++ b/api/base/hook/protopb/hook.pb.go @@ -0,0 +1,210 @@ +// 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. + +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.10 +// protoc v5.29.3 +// source: hook.proto + +package protopb + +import ( + reflect "reflect" + sync "sync" + unsafe "unsafe" + + _ "github.com/uber/submitqueue/api/base/messagequeue/protopb" + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + structpb "google.golang.org/protobuf/types/known/structpb" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// HookEvent is one fire-and-forget lifecycle event. Every domain publishes this +// same shape to its own hook topic, so a sink that consumes several domains +// reads one schema rather than one per producer. See api/base/hook/README.md. +type HookEvent struct { + state protoimpl.MessageState `protogen:"open.v1"` + // id is the opaque identity of this occurrence, derived from the transition + // it describes so that replaying the transition mints the same id. It is + // the queue's dedupe key and a hook's idempotency key, and is never parsed. + Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` + // source is the domain that produced the event: "submitqueue", + // "stovepipe", ... An open string rather than an enum so a new producer + // does not break existing consumers. + Source string `protobuf:"bytes,2,opt,name=source,proto3" json:"source,omitempty"` + // type is what happened, as one dotted open string: "request.landed", + // "batch.failed", ... It is the only dimension a consumer filters on, and + // open for the same reason as source. + Type string `protobuf:"bytes,3,opt,name=type,proto3" json:"type,omitempty"` + // timestamp_ms is when the occurrence happened, in milliseconds since the + // Unix epoch, on the publisher's clock. + TimestampMs int64 `protobuf:"varint,4,opt,name=timestamp_ms,json=timestampMs,proto3" json:"timestamp_ms,omitempty"` + // version is the subject's optimistic-locking version immediately after the + // transition, and 0 when the transition was not a versioned write. A hook + // compares it against the subject's current version to tell a current event + // from one that has since been superseded. + Version int32 `protobuf:"varint,5,opt,name=version,proto3" json:"version,omitempty"` + // payload carries the facts specific to type, always including the + // subject's id, and must carry any fact recorded nowhere else because the + // event is that fact's only durable record. Add-only, and never an entity + // snapshot: hooks resolve entities from their stores. + Payload *structpb.Struct `protobuf:"bytes,6,opt,name=payload,proto3" json:"payload,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *HookEvent) Reset() { + *x = HookEvent{} + mi := &file_hook_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *HookEvent) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HookEvent) ProtoMessage() {} + +func (x *HookEvent) ProtoReflect() protoreflect.Message { + mi := &file_hook_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HookEvent.ProtoReflect.Descriptor instead. +func (*HookEvent) Descriptor() ([]byte, []int) { + return file_hook_proto_rawDescGZIP(), []int{0} +} + +func (x *HookEvent) GetId() string { + if x != nil { + return x.Id + } + return "" +} + +func (x *HookEvent) GetSource() string { + if x != nil { + return x.Source + } + return "" +} + +func (x *HookEvent) GetType() string { + if x != nil { + return x.Type + } + return "" +} + +func (x *HookEvent) GetTimestampMs() int64 { + if x != nil { + return x.TimestampMs + } + return 0 +} + +func (x *HookEvent) GetVersion() int32 { + if x != nil { + return x.Version + } + return 0 +} + +func (x *HookEvent) GetPayload() *structpb.Struct { + if x != nil { + return x.Payload + } + return nil +} + +var File_hook_proto protoreflect.FileDescriptor + +const file_hook_proto_rawDesc = "" + + "\n" + + "\n" + + "hook.proto\x12\x0euber.base.hook\x1a\x1cgoogle/protobuf/struct.proto\x1a.api/base/messagequeue/proto/messagequeue.proto\"\xc1\x01\n" + + "\tHookEvent\x12\x0e\n" + + "\x02id\x18\x01 \x01(\tR\x02id\x12\x16\n" + + "\x06source\x18\x02 \x01(\tR\x06source\x12\x12\n" + + "\x04type\x18\x03 \x01(\tR\x04type\x12!\n" + + "\ftimestamp_ms\x18\x04 \x01(\x03R\vtimestampMs\x12\x18\n" + + "\aversion\x18\x05 \x01(\x05R\aversion\x121\n" + + "\apayload\x18\x06 \x01(\v2\x17.google.protobuf.StructR\apayload:\b\x8a\xb5\x18\x04hookB`\n" + + "\x1ecom.uber.submitqueue.base.hookB\tHookProtoP\x01Z1github.com/uber/submitqueue/api/base/hook/protopbb\x06proto3" + +var ( + file_hook_proto_rawDescOnce sync.Once + file_hook_proto_rawDescData []byte +) + +func file_hook_proto_rawDescGZIP() []byte { + file_hook_proto_rawDescOnce.Do(func() { + file_hook_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_hook_proto_rawDesc), len(file_hook_proto_rawDesc))) + }) + return file_hook_proto_rawDescData +} + +var file_hook_proto_msgTypes = make([]protoimpl.MessageInfo, 1) +var file_hook_proto_goTypes = []any{ + (*HookEvent)(nil), // 0: uber.base.hook.HookEvent + (*structpb.Struct)(nil), // 1: google.protobuf.Struct +} +var file_hook_proto_depIdxs = []int32{ + 1, // 0: uber.base.hook.HookEvent.payload:type_name -> google.protobuf.Struct + 1, // [1:1] is the sub-list for method output_type + 1, // [1:1] is the sub-list for method input_type + 1, // [1:1] is the sub-list for extension type_name + 1, // [1:1] is the sub-list for extension extendee + 0, // [0:1] is the sub-list for field type_name +} + +func init() { file_hook_proto_init() } +func file_hook_proto_init() { + if File_hook_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_hook_proto_rawDesc), len(file_hook_proto_rawDesc)), + NumEnums: 0, + NumMessages: 1, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_hook_proto_goTypes, + DependencyIndexes: file_hook_proto_depIdxs, + MessageInfos: file_hook_proto_msgTypes, + }.Build() + File_hook_proto = out.File + file_hook_proto_goTypes = nil + file_hook_proto_depIdxs = nil +} diff --git a/api/base/hook/topics.go b/api/base/hook/topics.go new file mode 100644 index 00000000..8640df72 --- /dev/null +++ b/api/base/hook/topics.go @@ -0,0 +1,26 @@ +// 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 + +import "github.com/uber/submitqueue/platform/consumer" + +// TopicKey looks up a queue backend, topic name, and subscription config in a +// consumer.TopicRegistry. +type TopicKey = consumer.TopicKey + +// TopicKeyHook carries hook events. The key is per-host, not global: each domain +// runs its own hook topic, so two domains sharing a queue backend must map this +// to distinct topic names. +const TopicKeyHook TopicKey = "hook" diff --git a/tool/proto/BUILD.bazel b/tool/proto/BUILD.bazel index 5b9ebca1..16bc485c 100644 --- a/tool/proto/BUILD.bazel +++ b/tool/proto/BUILD.bazel @@ -23,6 +23,16 @@ go_proto_generated_files( out_dir = "api_base_messagequeue", ) +go_proto_generated_files( + name = "api_base_hook", + srcs = ["//api/base/hook/proto:hook.proto"], + gen_services = False, + imports = [ + "//api/base/messagequeue/proto:messagequeue.proto", + ], + out_dir = "api_base_hook", +) + go_proto_generated_files( name = "api_runway_messagequeue", srcs = ["//api/runway/messagequeue/proto:merge.proto"], @@ -83,6 +93,7 @@ filegroup( name = "generated", srcs = [ ":api_base_change", + ":api_base_hook", ":api_base_mergestrategy", ":api_base_messagequeue", ":api_runway",