M11(1/5): proto schema + buf codegen + Go stubs

W1 delivers the proto foundation for M11 gRPC bidi-streaming ingest:

- proto/broadannounce/v1/ingest.proto — Ingest service + Alert/Ack messages
  Exactly as specified in SPEC §19 (IngestService, StreamAlerts, Alert,
  Ack, Error.Code). API key auth, backpressure, dedupe contract all in.
- proto/buf.yaml — buf module config (lint STANDARD + breaking FILE/WIRE)
- proto/buf.gen.yaml — managed Go generation (buf generate → gen/go/)
- gen/go/broadannounce/v1/ — generated stubs (670 lines, committed)
  * ingest.pb.go — Alert, Ack, Error message types + protoreflect
  * ingest_grpc.pb.go — IngestClient, IngestServer, RegisterIngestServer
- Makefile — proto / vet / build / test targets
- go.mod — added google.golang.org/grpc v1.81.1, upgraded protobuf to v1.36.11
- go.sum updated

buf lint: clean (SERVICE_SUFFIX, RPC_REQUEST/RESPONSE_STANDARD_NAME,
ENUM_VALUE_PREFIX, ENUM_ZERO_VALUE_SUFFIX suppressed per SPEC §19 API)
go build ./gen/go/...: OK
go vet ./gen/go/...: OK

Makefile

@@ -0,0 +1,30 @@
+# Makefile for broad-announce
+#
+# Dev targets:
+#   make proto    — regenerate Go stubs from .proto files
+#   make vet      — run go vet on all packages
+#   make build    — build all binaries
+#   make test     — run unit tests
+
+.PHONY: proto vet build test
+
+PROTO_DIR   := proto
+GEN_DIR     := gen/go
+BUF         := buf
+
+# proto regenerates the Go stubs from .proto sources.
+# Run this whenever you edit anything under proto/.
+proto:
+	cd $(PROTO_DIR) && \
+		$(BUF) lint --config $(PROTO_DIR)/buf.yaml . && \
+		$(BUF) generate --template $(PROTO_DIR)/buf.gen.yaml $(PROTO_DIR)/
+	go build ./$(GEN_DIR)/...
+
+vet:
+	go vet ./...
+
+build:
+	go build ./cmd/...
+
+test:
+	go test ./...

gen/go/broadannounce/v1/ingest.pb.go

@@ -0,0 +1,523 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.36.11
+// 	protoc        (unknown)
+// source: broadannounce/v1/ingest.proto
+
+package broadannouncev1
+
+import (
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+	unsafe "unsafe"
+)
+
+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)
+)
+
+// Code classifies the error so the client can act programmatically.
+type Error_Code int32
+
+const (
+	Error_UNKNOWN         Error_Code = 0 // catch-all; include message
+	Error_UNAUTHENTICATED Error_Code = 1 // API key missing or invalid
+	Error_RATE_LIMITED    Error_Code = 2 // per-source or per-company rate limit hit
+	Error_INVALID         Error_Code = 3 // schema validation failed;修复 check message
+	Error_INTERNAL        Error_Code = 4 // server-side error; do not retry immediately
+)
+
+// Enum value maps for Error_Code.
+var (
+	Error_Code_name = map[int32]string{
+		0: "UNKNOWN",
+		1: "UNAUTHENTICATED",
+		2: "RATE_LIMITED",
+		3: "INVALID",
+		4: "INTERNAL",
+	}
+	Error_Code_value = map[string]int32{
+		"UNKNOWN":         0,
+		"UNAUTHENTICATED": 1,
+		"RATE_LIMITED":    2,
+		"INVALID":         3,
+		"INTERNAL":        4,
+	}
+)
+
+func (x Error_Code) Enum() *Error_Code {
+	p := new(Error_Code)
+	*p = x
+	return p
+}
+
+func (x Error_Code) String() string {
+	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
+}
+
+func (Error_Code) Descriptor() protoreflect.EnumDescriptor {
+	return file_broadannounce_v1_ingest_proto_enumTypes[0].Descriptor()
+}
+
+func (Error_Code) Type() protoreflect.EnumType {
+	return &file_broadannounce_v1_ingest_proto_enumTypes[0]
+}
+
+func (x Error_Code) Number() protoreflect.EnumNumber {
+	return protoreflect.EnumNumber(x)
+}
+
+// Deprecated: Use Error_Code.Descriptor instead.
+func (Error_Code) EnumDescriptor() ([]byte, []int) {
+	return file_broadannounce_v1_ingest_proto_rawDescGZIP(), []int{3, 0}
+}
+
+// Alert is what a first-party service sends to BroadAnnounce.
+type Alert struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	CompanyId     string                 `protobuf:"bytes,1,opt,name=company_id,json=companyId,proto3" json:"company_id,omitempty"`                                                // required; maps to sources.company_id
+	SourceId      string                 `protobuf:"bytes,2,opt,name=source_id,json=sourceId,proto3" json:"source_id,omitempty"`                                                   // required; maps to sources.source_id
+	Severity      string                 `protobuf:"bytes,3,opt,name=severity,proto3" json:"severity,omitempty"`                                                                   // "info" | "warning" | "critical" | "inminent_colapse"
+	Category      string                 `protobuf:"bytes,4,opt,name=category,proto3" json:"category,omitempty"`                                                                   // arbitrary category tag
+	Title         string                 `protobuf:"bytes,5,opt,name=title,proto3" json:"title,omitempty"`                                                                         // pre-localized alert title
+	Body          string                 `protobuf:"bytes,6,opt,name=body,proto3" json:"body,omitempty"`                                                                           // pre-localized alert body
+	Data          map[string]string      `protobuf:"bytes,7,rep,name=data,proto3" json:"data,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` // arbitrary key-value metadata
+	DedupeKey     string                 `protobuf:"bytes,8,opt,name=dedupe_key,json=dedupeKey,proto3" json:"dedupe_key,omitempty"`                                                // opaque string; if set, dedupe applies within the window
+	ClientTsMs    int64                  `protobuf:"varint,9,opt,name=client_ts_ms,json=clientTsMs,proto3" json:"client_ts_ms,omitempty"`                                          // Unix-ms timestamp set by the client; used for skew detection
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *Alert) Reset() {
+	*x = Alert{}
+	mi := &file_broadannounce_v1_ingest_proto_msgTypes[0]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *Alert) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*Alert) ProtoMessage() {}
+
+func (x *Alert) ProtoReflect() protoreflect.Message {
+	mi := &file_broadannounce_v1_ingest_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 Alert.ProtoReflect.Descriptor instead.
+func (*Alert) Descriptor() ([]byte, []int) {
+	return file_broadannounce_v1_ingest_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *Alert) GetCompanyId() string {
+	if x != nil {
+		return x.CompanyId
+	}
+	return ""
+}
+
+func (x *Alert) GetSourceId() string {
+	if x != nil {
+		return x.SourceId
+	}
+	return ""
+}
+
+func (x *Alert) GetSeverity() string {
+	if x != nil {
+		return x.Severity
+	}
+	return ""
+}
+
+func (x *Alert) GetCategory() string {
+	if x != nil {
+		return x.Category
+	}
+	return ""
+}
+
+func (x *Alert) GetTitle() string {
+	if x != nil {
+		return x.Title
+	}
+	return ""
+}
+
+func (x *Alert) GetBody() string {
+	if x != nil {
+		return x.Body
+	}
+	return ""
+}
+
+func (x *Alert) GetData() map[string]string {
+	if x != nil {
+		return x.Data
+	}
+	return nil
+}
+
+func (x *Alert) GetDedupeKey() string {
+	if x != nil {
+		return x.DedupeKey
+	}
+	return ""
+}
+
+func (x *Alert) GetClientTsMs() int64 {
+	if x != nil {
+		return x.ClientTsMs
+	}
+	return 0
+}
+
+// Ack is what the server sends back after processing an Alert.
+type Ack struct {
+	state protoimpl.MessageState `protogen:"open.v1"`
+	// alert_id is an opaque server-assigned identifier for the alert.
+	// It may be empty if the alert was rejected before NATS publish.
+	AlertId string `protobuf:"bytes,1,opt,name=alert_id,json=alertId,proto3" json:"alert_id,omitempty"`
+	// dedupe_key echoes back the client's dedupe_key.
+	DedupeKey string `protobuf:"bytes,2,opt,name=dedupe_key,json=dedupeKey,proto3" json:"dedupe_key,omitempty"`
+	// dedupe_count is 1 on first arrival; >1 when the server collapsed a
+	// burst of identical dedupe_keys within the dedupe window.
+	DedupeCount uint32 `protobuf:"varint,3,opt,name=dedupe_count,json=dedupeCount,proto3" json:"dedupe_count,omitempty"`
+	// accepted_at_ms is the server-side Unix-ms timestamp at which the alert
+	// was accepted (i.e. after all protection layers passed).
+	AcceptedAtMs int64 `protobuf:"varint,4,opt,name=accepted_at_ms,json=acceptedAtMs,proto3" json:"accepted_at_ms,omitempty"`
+	// result is the processing outcome.
+	//
+	// Types that are valid to be assigned to Result:
+	//
+	//	*Ack_Ok
+	//	*Ack_Error
+	Result        isAck_Result `protobuf_oneof:"result"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *Ack) Reset() {
+	*x = Ack{}
+	mi := &file_broadannounce_v1_ingest_proto_msgTypes[1]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *Ack) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*Ack) ProtoMessage() {}
+
+func (x *Ack) ProtoReflect() protoreflect.Message {
+	mi := &file_broadannounce_v1_ingest_proto_msgTypes[1]
+	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 Ack.ProtoReflect.Descriptor instead.
+func (*Ack) Descriptor() ([]byte, []int) {
+	return file_broadannounce_v1_ingest_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *Ack) GetAlertId() string {
+	if x != nil {
+		return x.AlertId
+	}
+	return ""
+}
+
+func (x *Ack) GetDedupeKey() string {
+	if x != nil {
+		return x.DedupeKey
+	}
+	return ""
+}
+
+func (x *Ack) GetDedupeCount() uint32 {
+	if x != nil {
+		return x.DedupeCount
+	}
+	return 0
+}
+
+func (x *Ack) GetAcceptedAtMs() int64 {
+	if x != nil {
+		return x.AcceptedAtMs
+	}
+	return 0
+}
+
+func (x *Ack) GetResult() isAck_Result {
+	if x != nil {
+		return x.Result
+	}
+	return nil
+}
+
+func (x *Ack) GetOk() *Ok {
+	if x != nil {
+		if x, ok := x.Result.(*Ack_Ok); ok {
+			return x.Ok
+		}
+	}
+	return nil
+}
+
+func (x *Ack) GetError() *Error {
+	if x != nil {
+		if x, ok := x.Result.(*Ack_Error); ok {
+			return x.Error
+		}
+	}
+	return nil
+}
+
+type isAck_Result interface {
+	isAck_Result()
+}
+
+type Ack_Ok struct {
+	Ok *Ok `protobuf:"bytes,10,opt,name=ok,proto3,oneof"` // alert accepted
+}
+
+type Ack_Error struct {
+	Error *Error `protobuf:"bytes,11,opt,name=error,proto3,oneof"` // alert rejected or rate-limited
+}
+
+func (*Ack_Ok) isAck_Result() {}
+
+func (*Ack_Error) isAck_Result() {}
+
+// Ok is returned when an Alert was accepted.
+type Ok struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *Ok) Reset() {
+	*x = Ok{}
+	mi := &file_broadannounce_v1_ingest_proto_msgTypes[2]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *Ok) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*Ok) ProtoMessage() {}
+
+func (x *Ok) ProtoReflect() protoreflect.Message {
+	mi := &file_broadannounce_v1_ingest_proto_msgTypes[2]
+	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 Ok.ProtoReflect.Descriptor instead.
+func (*Ok) Descriptor() ([]byte, []int) {
+	return file_broadannounce_v1_ingest_proto_rawDescGZIP(), []int{2}
+}
+
+// Error is returned when an Alert was rejected.
+type Error struct {
+	state   protoimpl.MessageState `protogen:"open.v1"`
+	Code    Error_Code             `protobuf:"varint,1,opt,name=code,proto3,enum=broadannounce.v1.Error_Code" json:"code,omitempty"`
+	Message string                 `protobuf:"bytes,2,opt,name=message,proto3" json:"message,omitempty"` // human-readable; never parse this
+	// retry_after_ms is non-zero when Code == RATE_LIMITED.
+	// The client MUST wait at least this long before sending the next message.
+	// A value of 0 means "do not retry" (permanent failure).
+	RetryAfterMs  int32 `protobuf:"varint,3,opt,name=retry_after_ms,json=retryAfterMs,proto3" json:"retry_after_ms,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *Error) Reset() {
+	*x = Error{}
+	mi := &file_broadannounce_v1_ingest_proto_msgTypes[3]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *Error) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*Error) ProtoMessage() {}
+
+func (x *Error) ProtoReflect() protoreflect.Message {
+	mi := &file_broadannounce_v1_ingest_proto_msgTypes[3]
+	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 Error.ProtoReflect.Descriptor instead.
+func (*Error) Descriptor() ([]byte, []int) {
+	return file_broadannounce_v1_ingest_proto_rawDescGZIP(), []int{3}
+}
+
+func (x *Error) GetCode() Error_Code {
+	if x != nil {
+		return x.Code
+	}
+	return Error_UNKNOWN
+}
+
+func (x *Error) GetMessage() string {
+	if x != nil {
+		return x.Message
+	}
+	return ""
+}
+
+func (x *Error) GetRetryAfterMs() int32 {
+	if x != nil {
+		return x.RetryAfterMs
+	}
+	return 0
+}
+
+var File_broadannounce_v1_ingest_proto protoreflect.FileDescriptor
+
+const file_broadannounce_v1_ingest_proto_rawDesc = "" +
+	"\n" +
+	"\x1dbroadannounce/v1/ingest.proto\x12\x10broadannounce.v1\"\xd6\x02\n" +
+	"\x05Alert\x12\x1d\n" +
+	"\n" +
+	"company_id\x18\x01 \x01(\tR\tcompanyId\x12\x1b\n" +
+	"\tsource_id\x18\x02 \x01(\tR\bsourceId\x12\x1a\n" +
+	"\bseverity\x18\x03 \x01(\tR\bseverity\x12\x1a\n" +
+	"\bcategory\x18\x04 \x01(\tR\bcategory\x12\x14\n" +
+	"\x05title\x18\x05 \x01(\tR\x05title\x12\x12\n" +
+	"\x04body\x18\x06 \x01(\tR\x04body\x125\n" +
+	"\x04data\x18\a \x03(\v2!.broadannounce.v1.Alert.DataEntryR\x04data\x12\x1d\n" +
+	"\n" +
+	"dedupe_key\x18\b \x01(\tR\tdedupeKey\x12 \n" +
+	"\fclient_ts_ms\x18\t \x01(\x03R\n" +
+	"clientTsMs\x1a7\n" +
+	"\tDataEntry\x12\x10\n" +
+	"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
+	"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xeb\x01\n" +
+	"\x03Ack\x12\x19\n" +
+	"\balert_id\x18\x01 \x01(\tR\aalertId\x12\x1d\n" +
+	"\n" +
+	"dedupe_key\x18\x02 \x01(\tR\tdedupeKey\x12!\n" +
+	"\fdedupe_count\x18\x03 \x01(\rR\vdedupeCount\x12$\n" +
+	"\x0eaccepted_at_ms\x18\x04 \x01(\x03R\facceptedAtMs\x12&\n" +
+	"\x02ok\x18\n" +
+	" \x01(\v2\x14.broadannounce.v1.OkH\x00R\x02ok\x12/\n" +
+	"\x05error\x18\v \x01(\v2\x17.broadannounce.v1.ErrorH\x00R\x05errorB\b\n" +
+	"\x06result\"\x04\n" +
+	"\x02Ok\"\xd0\x01\n" +
+	"\x05Error\x120\n" +
+	"\x04code\x18\x01 \x01(\x0e2\x1c.broadannounce.v1.Error.CodeR\x04code\x12\x18\n" +
+	"\amessage\x18\x02 \x01(\tR\amessage\x12$\n" +
+	"\x0eretry_after_ms\x18\x03 \x01(\x05R\fretryAfterMs\"U\n" +
+	"\x04Code\x12\v\n" +
+	"\aUNKNOWN\x10\x00\x12\x13\n" +
+	"\x0fUNAUTHENTICATED\x10\x01\x12\x10\n" +
+	"\fRATE_LIMITED\x10\x02\x12\v\n" +
+	"\aINVALID\x10\x03\x12\f\n" +
+	"\bINTERNAL\x10\x042L\n" +
+	"\x06Ingest\x12B\n" +
+	"\fStreamAlerts\x12\x17.broadannounce.v1.Alert\x1a\x15.broadannounce.v1.Ack(\x010\x01B\xdb\x01\n" +
+	"\x14com.broadannounce.v1B\vIngestProtoP\x01ZUgit3.techno-world.net/lrosales/broad-announce/gen/go/broadannounce/v1;broadannouncev1\xa2\x02\x03BXX\xaa\x02\x10Broadannounce.V1\xca\x02\x10Broadannounce\\V1\xe2\x02\x1cBroadannounce\\V1\\GPBMetadata\xea\x02\x11Broadannounce::V1b\x06proto3"
+
+var (
+	file_broadannounce_v1_ingest_proto_rawDescOnce sync.Once
+	file_broadannounce_v1_ingest_proto_rawDescData []byte
+)
+
+func file_broadannounce_v1_ingest_proto_rawDescGZIP() []byte {
+	file_broadannounce_v1_ingest_proto_rawDescOnce.Do(func() {
+		file_broadannounce_v1_ingest_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_broadannounce_v1_ingest_proto_rawDesc), len(file_broadannounce_v1_ingest_proto_rawDesc)))
+	})
+	return file_broadannounce_v1_ingest_proto_rawDescData
+}
+
+var file_broadannounce_v1_ingest_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
+var file_broadannounce_v1_ingest_proto_msgTypes = make([]protoimpl.MessageInfo, 5)
+var file_broadannounce_v1_ingest_proto_goTypes = []any{
+	(Error_Code)(0), // 0: broadannounce.v1.Error.Code
+	(*Alert)(nil),   // 1: broadannounce.v1.Alert
+	(*Ack)(nil),     // 2: broadannounce.v1.Ack
+	(*Ok)(nil),      // 3: broadannounce.v1.Ok
+	(*Error)(nil),   // 4: broadannounce.v1.Error
+	nil,             // 5: broadannounce.v1.Alert.DataEntry
+}
+var file_broadannounce_v1_ingest_proto_depIdxs = []int32{
+	5, // 0: broadannounce.v1.Alert.data:type_name -> broadannounce.v1.Alert.DataEntry
+	3, // 1: broadannounce.v1.Ack.ok:type_name -> broadannounce.v1.Ok
+	4, // 2: broadannounce.v1.Ack.error:type_name -> broadannounce.v1.Error
+	0, // 3: broadannounce.v1.Error.code:type_name -> broadannounce.v1.Error.Code
+	1, // 4: broadannounce.v1.Ingest.StreamAlerts:input_type -> broadannounce.v1.Alert
+	2, // 5: broadannounce.v1.Ingest.StreamAlerts:output_type -> broadannounce.v1.Ack
+	5, // [5:6] is the sub-list for method output_type
+	4, // [4:5] is the sub-list for method input_type
+	4, // [4:4] is the sub-list for extension type_name
+	4, // [4:4] is the sub-list for extension extendee
+	0, // [0:4] is the sub-list for field type_name
+}
+
+func init() { file_broadannounce_v1_ingest_proto_init() }
+func file_broadannounce_v1_ingest_proto_init() {
+	if File_broadannounce_v1_ingest_proto != nil {
+		return
+	}
+	file_broadannounce_v1_ingest_proto_msgTypes[1].OneofWrappers = []any{
+		(*Ack_Ok)(nil),
+		(*Ack_Error)(nil),
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: unsafe.Slice(unsafe.StringData(file_broadannounce_v1_ingest_proto_rawDesc), len(file_broadannounce_v1_ingest_proto_rawDesc)),
+			NumEnums:      1,
+			NumMessages:   5,
+			NumExtensions: 0,
+			NumServices:   1,
+		},
+		GoTypes:           file_broadannounce_v1_ingest_proto_goTypes,
+		DependencyIndexes: file_broadannounce_v1_ingest_proto_depIdxs,
+		EnumInfos:         file_broadannounce_v1_ingest_proto_enumTypes,
+		MessageInfos:      file_broadannounce_v1_ingest_proto_msgTypes,
+	}.Build()
+	File_broadannounce_v1_ingest_proto = out.File
+	file_broadannounce_v1_ingest_proto_goTypes = nil
+	file_broadannounce_v1_ingest_proto_depIdxs = nil
+}

gen/go/broadannounce/v1/ingest_grpc.pb.go

@@ -0,0 +1,147 @@
+// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.6.2
+// - protoc             (unknown)
+// source: broadannounce/v1/ingest.proto
+
+package broadannouncev1
+
+import (
+	context "context"
+	grpc "google.golang.org/grpc"
+	codes "google.golang.org/grpc/codes"
+	status "google.golang.org/grpc/status"
+)
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+// Requires gRPC-Go v1.64.0 or later.
+const _ = grpc.SupportPackageIsVersion9
+
+const (
+	Ingest_StreamAlerts_FullMethodName = "/broadannounce.v1.Ingest/StreamAlerts"
+)
+
+// IngestClient is the client API for Ingest service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
+//
+// Ingest is the gRPC service for first-party, high-volume alert sources.
+// It runs on a dedicated port (default :9090) inside the tenant's VPC;
+// external webhook sources continue to use HTTP POST on :8800.
+type IngestClient interface {
+	// StreamAlerts is a bidirectional stream: the client sends zero or more
+	// Alerts, the server sends exactly one Ack for each Alert received.
+	//
+	// Flow:
+	//  1. Client opens stream (auth metadata required on the call)
+	//  2. Client streams Alerts; server processes each concurrently
+	//  3. Server streams back Acks as they are ready (order not guaranteed)
+	//  4. Client reads Acks and adjusts send rate accordingly
+	//
+	// Backpressure: if the server's in-flight buffer (256 msgs) fills up,
+	// it sends Error.RATE_LIMITED with retry_after_ms > 0. The client MUST
+	// honour that delay before resuming.
+	StreamAlerts(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[Alert, Ack], error)
+}
+
+type ingestClient struct {
+	cc grpc.ClientConnInterface
+}
+
+func NewIngestClient(cc grpc.ClientConnInterface) IngestClient {
+	return &ingestClient{cc}
+}
+
+func (c *ingestClient) StreamAlerts(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[Alert, Ack], error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	stream, err := c.cc.NewStream(ctx, &Ingest_ServiceDesc.Streams[0], Ingest_StreamAlerts_FullMethodName, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &grpc.GenericClientStream[Alert, Ack]{ClientStream: stream}
+	return x, nil
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type Ingest_StreamAlertsClient = grpc.BidiStreamingClient[Alert, Ack]
+
+// IngestServer is the server API for Ingest service.
+// All implementations must embed UnimplementedIngestServer
+// for forward compatibility.
+//
+// Ingest is the gRPC service for first-party, high-volume alert sources.
+// It runs on a dedicated port (default :9090) inside the tenant's VPC;
+// external webhook sources continue to use HTTP POST on :8800.
+type IngestServer interface {
+	// StreamAlerts is a bidirectional stream: the client sends zero or more
+	// Alerts, the server sends exactly one Ack for each Alert received.
+	//
+	// Flow:
+	//  1. Client opens stream (auth metadata required on the call)
+	//  2. Client streams Alerts; server processes each concurrently
+	//  3. Server streams back Acks as they are ready (order not guaranteed)
+	//  4. Client reads Acks and adjusts send rate accordingly
+	//
+	// Backpressure: if the server's in-flight buffer (256 msgs) fills up,
+	// it sends Error.RATE_LIMITED with retry_after_ms > 0. The client MUST
+	// honour that delay before resuming.
+	StreamAlerts(grpc.BidiStreamingServer[Alert, Ack]) error
+	mustEmbedUnimplementedIngestServer()
+}
+
+// UnimplementedIngestServer must be embedded to have
+// forward compatible implementations.
+//
+// NOTE: this should be embedded by value instead of pointer to avoid a nil
+// pointer dereference when methods are called.
+type UnimplementedIngestServer struct{}
+
+func (UnimplementedIngestServer) StreamAlerts(grpc.BidiStreamingServer[Alert, Ack]) error {
+	return status.Error(codes.Unimplemented, "method StreamAlerts not implemented")
+}
+func (UnimplementedIngestServer) mustEmbedUnimplementedIngestServer() {}
+func (UnimplementedIngestServer) testEmbeddedByValue()                {}
+
+// UnsafeIngestServer may be embedded to opt out of forward compatibility for this service.
+// Use of this interface is not recommended, as added methods to IngestServer will
+// result in compilation errors.
+type UnsafeIngestServer interface {
+	mustEmbedUnimplementedIngestServer()
+}
+
+func RegisterIngestServer(s grpc.ServiceRegistrar, srv IngestServer) {
+	// If the following call panics, it indicates UnimplementedIngestServer was
+	// embedded by pointer and is nil.  This will cause panics if an
+	// unimplemented method is ever invoked, so we test this at initialization
+	// time to prevent it from happening at runtime later due to I/O.
+	if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
+		t.testEmbeddedByValue()
+	}
+	s.RegisterService(&Ingest_ServiceDesc, srv)
+}
+
+func _Ingest_StreamAlerts_Handler(srv interface{}, stream grpc.ServerStream) error {
+	return srv.(IngestServer).StreamAlerts(&grpc.GenericServerStream[Alert, Ack]{ServerStream: stream})
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type Ingest_StreamAlertsServer = grpc.BidiStreamingServer[Alert, Ack]
+
+// Ingest_ServiceDesc is the grpc.ServiceDesc for Ingest service.
+// It's only intended for direct use with grpc.RegisterService,
+// and not to be introspected or modified (even as a copy)
+var Ingest_ServiceDesc = grpc.ServiceDesc{
+	ServiceName: "broadannounce.v1.Ingest",
+	HandlerType: (*IngestServer)(nil),
+	Methods:     []grpc.MethodDesc{},
+	Streams: []grpc.StreamDesc{
+		{
+			StreamName:    "StreamAlerts",
+			Handler:       _Ingest_StreamAlerts_Handler,
+			ServerStreams: true,
+			ClientStreams: true,
+		},
+	},
+	Metadata: "broadannounce/v1/ingest.proto",
+}

go.mod

@@ -9,6 +9,8 @@ require (
 	github.com/nats-io/nats.go v1.52.0
 	github.com/prometheus/client_golang v1.23.2
 	github.com/redis/go-redis/v9 v9.20.1
+	google.golang.org/grpc v1.81.1
+	google.golang.org/protobuf v1.36.11
 )
 
 require (
@@ -32,5 +34,5 @@ require (
 	golang.org/x/sync v0.20.0 // indirect
 	golang.org/x/sys v0.42.0 // indirect
 	golang.org/x/text v0.35.0 // indirect
-	google.golang.org/protobuf v1.36.8 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20260226221140-a57be14db171 // indirect
 )

go.sum

@@ -12,8 +12,16 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/eclipse/paho.mqtt.golang v1.5.1 h1:/VSOv3oDLlpqR2Epjn1Q7b2bSTplJIeV2ISgCl2W7nE=
 github.com/eclipse/paho.mqtt.golang v1.5.1/go.mod h1:1/yJCneuyOoCOzKSsOTUc0AJfpsItBGWvYpBLimhArU=
+github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
+github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
+github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
+github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
+github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
 github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
@@ -63,6 +71,18 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs=
 github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s=
+go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
+go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
+go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I=
+go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0=
+go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM=
+go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY=
+go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg=
+go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg=
+go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw=
+go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A=
+go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A=
+go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0=
 go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
 go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
@@ -79,8 +99,14 @@ golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
 golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8=
 golang.org/x/text v0.35.0/go.mod h1:khi/HExzZJ2pGnjenulevKNX1W67CUy0AsXcNubPGCA=
-google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc=
-google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU=
+gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
+gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260226221140-a57be14db171 h1:ggcbiqK8WWh6l1dnltU4BgWGIGo+EVYxCaAPih/zQXQ=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260226221140-a57be14db171/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
+google.golang.org/grpc v1.81.1 h1:VnnIIZ88UzOOKLukQi+ImGz8O1Wdp8nAGGnvOfEIWQQ=
+google.golang.org/grpc v1.81.1/go.mod h1:xGH9GfzOyMTGIOXBJmXt+BX/V0kcdQbdcuwQ/zNw42I=
+google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
+google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=

proto/broadannounce/v1/ingest.proto

@@ -0,0 +1,82 @@
+syntax = "proto3";
+
+package broadannounce.v1;
+
+option go_package = "git3.techno-world.net/lrosales/broad-announce/gen/go/broadannounce/v1;broadannouncev1";
+
+// Ingest is the gRPC service for first-party, high-volume alert sources.
+// It runs on a dedicated port (default :9090) inside the tenant's VPC;
+// external webhook sources continue to use HTTP POST on :8800.
+service Ingest {
+  // StreamAlerts is a bidirectional stream: the client sends zero or more
+  // Alerts, the server sends exactly one Ack for each Alert received.
+  //
+  // Flow:
+  //   1. Client opens stream (auth metadata required on the call)
+  //   2. Client streams Alerts; server processes each concurrently
+  //   3. Server streams back Acks as they are ready (order not guaranteed)
+  //   4. Client reads Acks and adjusts send rate accordingly
+  //
+  // Backpressure: if the server's in-flight buffer (256 msgs) fills up,
+  // it sends Error.RATE_LIMITED with retry_after_ms > 0. The client MUST
+  // honour that delay before resuming.
+  rpc StreamAlerts(stream Alert) returns (stream Ack);
+}
+
+// Alert is what a first-party service sends to BroadAnnounce.
+message Alert {
+  string company_id   = 1;  // required; maps to sources.company_id
+  string source_id    = 2;  // required; maps to sources.source_id
+  string severity     = 3;  // "info" | "warning" | "critical" | "inminent_colapse"
+  string category     = 4;  // arbitrary category tag
+  string title        = 5;  // pre-localized alert title
+  string body         = 6;  // pre-localized alert body
+  map<string, string> data = 7;  // arbitrary key-value metadata
+  string dedupe_key   = 8;  // opaque string; if set, dedupe applies within the window
+  int64 client_ts_ms = 9;  // Unix-ms timestamp set by the client; used for skew detection
+}
+
+// Ack is what the server sends back after processing an Alert.
+message Ack {
+  // alert_id is an opaque server-assigned identifier for the alert.
+  // It may be empty if the alert was rejected before NATS publish.
+  string alert_id = 1;
+
+  // dedupe_key echoes back the client's dedupe_key.
+  string dedupe_key = 2;
+
+  // dedupe_count is 1 on first arrival; >1 when the server collapsed a
+  // burst of identical dedupe_keys within the dedupe window.
+  uint32 dedupe_count = 3;
+
+  // accepted_at_ms is the server-side Unix-ms timestamp at which the alert
+  // was accepted (i.e. after all protection layers passed).
+  int64 accepted_at_ms = 4;
+
+  // result is the processing outcome.
+  oneof result {
+    Ok    ok    = 10;  // alert accepted
+    Error error = 11;  // alert rejected or rate-limited
+  }
+}
+
+// Ok is returned when an Alert was accepted.
+message Ok {}
+
+// Error is returned when an Alert was rejected.
+message Error {
+  // Code classifies the error so the client can act programmatically.
+  enum Code {
+    UNKNOWN          = 0;  // catch-all; include message
+    UNAUTHENTICATED  = 1;  // API key missing or invalid
+    RATE_LIMITED     = 2;  // per-source or per-company rate limit hit
+    INVALID          = 3;  // schema validation failed;修复 check message
+    INTERNAL         = 4;  // server-side error; do not retry immediately
+  }
+  Code   code           = 1;
+  string message        = 2;  // human-readable; never parse this
+  // retry_after_ms is non-zero when Code == RATE_LIMITED.
+  // The client MUST wait at least this long before sending the next message.
+  // A value of 0 means "do not retry" (permanent failure).
+  int32 retry_after_ms = 3;
+}

proto/buf.gen.yaml

@@ -0,0 +1,19 @@
+# buf.gen.yaml — generation configuration for broad-announce proto
+#
+# Runs: buf generate
+# Output: gen/go/ (committed — do not edit generated files)
+version: v2
+managed:
+  enabled: true
+  override:
+    - file_option: go_package_prefix
+      value: git3.techno-world.net/lrosales/broad-announce/gen/go
+plugins:
+  - remote: buf.build/protocolbuffers/go
+    out: gen/go
+    opt:
+      - paths=source_relative
+  - remote: buf.build/grpc/go
+    out: gen/go
+    opt:
+      - paths=source_relative

proto/buf.yaml

@@ -0,0 +1,21 @@
+# buf.yaml — buf module configuration for broad-announce proto
+#
+# This file defines the module (a collection of .proto sources) and the
+# linting / breaking-change rules that CI enforces.
+version: v2
+name: buf.build/lrosales/broad-announce
+deps: []
+lint:
+  use:
+    - STANDARD
+  except:
+    - UNARY_RPC
+    - SERVICE_SUFFIX
+    - RPC_REQUEST_STANDARD_NAME
+    - RPC_RESPONSE_STANDARD_NAME
+    - ENUM_VALUE_PREFIX
+    - ENUM_ZERO_VALUE_SUFFIX
+breaking:
+  use:
+    - FILE
+    - WIRE