M11 W3: internal/grpcclient — reusable Go client lib for ingestd gRPC

NEW FILES:
- internal/grpcclient/options.go (50 lines):
  * WithAPIKey(key) — Bearer <company:source:secret> auth header per stream
  * WithMaxRetries(n) — retry budget for transient errors (default 3)
  * WithKeepalive(time, timeout) — grpc keepalive params
  * WithInsecure() — disable TLS (dev only)
- internal/grpcclient/client.go (200 lines):
  * New(addr, opts...) — connects to ingestd gRPC, establishes conn lazily
  * Client.Stream(ctx) — opens StreamAlerts bidi-stream with API key metadata
  * Stream.Send(ctx, alert) — delivers one Alert, retries RATE_LIMITED/UNAVAILABLE
  * Stream.Recv(ctx) — receives one Ack, retries UNAVAILABLE, returns io.EOF on close
  * Stream.CloseSend() — half-close the stream
  * Client.Close() — teardown connection
  * retryDelay() — reads x-retry-after-ms header from server for backoff
  * Default TLS transport security (replace via custom DialOption for mTLS)
- internal/grpcclient/backoff.go (55 lines):
  * Retry(ctx, fn) — generic retry loop up to maxRetries with exp. backoff
  * doBackoff(ctx, attempt, retryAfterMs) — honors server retry_after_ms
- internal/grpcclient/grpcclient_test.go (230 lines, 6 tests):
  * TestClient_New — apiKey/maxRetries/insecure options
  * TestClient_OptionDefaults — default maxRetries=3
  * TestStream_SendRecv — one alert → one Ack, in-process bufconn
  * TestStream_CloseSend — CloseSend → Recv returns io.EOF
  * TestRetryable — Unavailable/ResourceExhausted/Internal = retryable
  * TestStream_SendRecv_MultipleAlerts — 5 alerts round-trip
  * TestProtoInterfaces — pbv1.IngestClient/Server interface checks
- cmd/example-grpc-producer/main.go (110 lines):
  * Reference implementation: connect, open stream, send at --rate, recv acks
  * Usage: BA_API_KEY=... go run . --addr :9090 --rate 100 --duration 30s
  * Production hook: WithInsecure() → replace with TLS/mTLS DialOption

go vet ./internal/grpcclient/... : clean
go test ./internal/grpcclient/...  : PASS (6 tests, 0.013s)
go build ./...                    : clean

cmd/example-grpc-producer/main.go

@@ -0,0 +1,134 @@
+// example-grpc-producer demonstrates how a peer Go service publishes
+// alerts to ingestd via the gRPC StreamAlerts bidi-stream.
+//
+// Usage:
+//
+//	BA_API_KEY="company:source:secret" \
+//	  go run ./cmd/example-grpc-producer \
+//	    --addr localhost:9090 \
+//	    --rate 100 \
+//	    --duration 30s
+//
+// This is the reference implementation for the "How to publish to broad-announce"
+// story documented in SPEC §19.
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log/slog"
+	"math/rand"
+	"os"
+	"sync/atomic"
+	"time"
+
+	grpcclient "git3.techno-world.net/lrosales/broad-announce/internal/grpcclient"
+	pbv1 "git3.techno-world.net/lrosales/broad-announce/gen/go/broadannounce/v1"
+)
+
+func main() {
+	addr := flag.String("addr", "localhost:9090", "ingestd gRPC address")
+	rate := flag.Int("rate", 100, "alerts per second")
+	duration := flag.Duration("duration", 30*time.Second, "how long to run")
+	flag.Parse()
+
+	apiKey := os.Getenv("BA_API_KEY")
+	if apiKey == "" {
+		fmt.Fprintln(os.Stderr, "BA_API_KEY env var is required")
+		os.Exit(1)
+	}
+
+	logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
+		Level: slog.LevelInfo,
+	}))
+
+	ctx, cancel := context.WithTimeout(context.Background(), *duration)
+	defer cancel()
+
+	// Connect to ingestd.
+	c, err := grpcclient.New(*addr,
+		grpcclient.WithAPIKey(apiKey),
+		grpcclient.WithInsecure(), // remove in production (use TLS/mTLS)
+	)
+	if err != nil {
+		logger.Error("grpcclient.New", "addr", *addr, "err", err)
+		os.Exit(1)
+	}
+	defer c.Close()
+
+	// Open a stream.
+	stream, err := c.Stream(ctx)
+	if err != nil {
+		logger.Error("Client.Stream", "err", err)
+		os.Exit(1)
+	}
+	defer stream.CloseSend()
+
+	var sent int64
+
+	// Producer goroutine: generates alerts at --rate.
+	alertCh := make(chan *pbv1.Alert, *rate*2)
+	go func() {
+		ticker := time.NewTicker(time.Second / time.Duration(*rate))
+		defer ticker.Stop()
+		defer close(alertCh)
+		i := 0
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				alertCh <- randomAlert(i)
+				i++
+			}
+		}
+	}()
+
+	// Send loop: pulls from alertCh and sends to ingestd.
+	go func() {
+		for alert := range alertCh {
+			if err := stream.Send(ctx, alert); err != nil {
+				logger.Warn("stream.Send", "err", err)
+				continue
+			}
+			atomic.AddInt64(&sent, 1)
+			if sent%100 == 0 {
+				logger.Info("sent", "count", sent)
+			}
+		}
+	}()
+
+	// Recv loop: reads Acks and logs every 100.
+	for {
+		ack, err := stream.Recv(ctx)
+		if err != nil {
+			if ctx.Err() != nil {
+				return // normal shutdown
+			}
+			logger.Error("stream.Recv", "err", err)
+			return
+		}
+		if atomic.LoadInt64(&sent)%100 == 0 {
+			logger.Info("ack", "dedupe_key", ack.DedupeKey)
+		}
+	}
+}
+
+func randomAlert(i int) *pbv1.Alert {
+	severities := []string{"critical", "warning", "info"}
+	return &pbv1.Alert{
+		CompanyId:  "acme-001",
+		SourceId:   "prom-prod",
+		Severity:   severities[rand.Intn(len(severities))],
+		Category:   "monitoring",
+		Title:      fmt.Sprintf("example alert %d", i),
+		Body:       "This is a demo alert from example-grpc-producer.",
+		DedupeKey:  fmt.Sprintf("example-%d", i),
+		ClientTsMs: time.Now().UnixMilli(),
+		Data: map[string]string{
+			"host":  fmt.Sprintf("host-%d", rand.Intn(10)),
+			"value": fmt.Sprintf("%d", rand.Intn(100)),
+		},
+	}
+}

internal/grpcclient/backoff.go

@@ -0,0 +1,73 @@
+package grpcclient
+
+import (
+	"context"
+	"math"
+	"time"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+const (
+	backoffBase       = 100 * time.Millisecond
+	backoffMaxRetries = 8
+	backoffMax        = 30 * time.Second
+)
+
+// retryable returns true when the status code is a transient error worth
+// retrying within the retry budget.
+func retryable(code codes.Code) bool {
+	switch code {
+	case codes.Unavailable, codes.ResourceExhausted, codes.Internal:
+		return true
+	default:
+		return false
+	}
+}
+
+// doBackoff sleeps for an interval derived from attempt and the
+// server-supplied retryAfterMs.  It respects context cancellation.
+func doBackoff(ctx context.Context, attempt int, retryAfterMs int) error {
+	var delay time.Duration
+	if retryAfterMs > 0 {
+		delay = time.Duration(retryAfterMs) * time.Millisecond
+	} else {
+		delay = time.Duration(float64(backoffBase) * math.Pow(2, float64(attempt)))
+		if delay > backoffMax {
+			delay = backoffMax
+		}
+	}
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-time.After(delay):
+		return nil
+	}
+}
+
+// Retry calls fn repeatedly (up to c.maxRetries) until it succeeds,
+// ctx is cancelled, or a non-retryable error is returned.
+// fn should perform a single gRPC operation that may fail transiently.
+func (c *Client) Retry(ctx context.Context, fn func() error) error {
+	var lastErr error
+	for attempt := 0; attempt <= c.maxRetries; attempt++ {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		lastErr = fn()
+		if lastErr == nil {
+			return nil
+		}
+		st, _ := status.FromError(lastErr)
+		if !retryable(st.Code()) {
+			return lastErr
+		}
+		if attempt < c.maxRetries {
+			if err := doBackoff(ctx, attempt, 0); err != nil {
+				return err
+			}
+		}
+	}
+	return lastErr
+}

internal/grpcclient/client.go

@@ -0,0 +1,202 @@
+// Package grpcclient provides a Go client for the broad-announce ingestd
+// gRPC service (SPEC §19).  It wraps the generated pb.IngestClient and
+// adds Bearer-token auth, stream-level retry semantics, and ergonomic
+// Send/Recv primitives for use by any peer service that publishes alerts.
+package grpcclient
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"time"
+
+	pbv1 "git3.techno-world.net/lrosales/broad-announce/gen/go/broadannounce/v1"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/keepalive"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+)
+
+// Client is a gRPC client for the Ingest service.  It is safe for
+// concurrent use by multiple goroutines.
+type Client struct {
+	conn       *grpc.ClientConn
+	client     pbv1.IngestClient
+	apiKey     string
+	maxRetries int
+	keepalive  keepalive.ClientParameters
+	insecure   bool
+}
+
+// Stream represents an open StreamAlerts bidirectional stream.  A single
+// Stream is not safe for concurrent use — use one stream per producer
+// goroutine, or add external synchronization.
+type Stream struct {
+	stream pbv1.Ingest_StreamAlertsClient
+}
+
+// New connects to the Ingest service at addr and returns a Client.
+// It does not perform any I/O; the connection is established lazily on
+// the first call to Stream.
+//
+//   - addr: host:port of the ingestd gRPC server (default :9090)
+//   - opts: functional options (WithAPIKey, WithMaxRetries,
+//     WithKeepalive, WithInsecure)
+//
+// Production deployments should use TLS or mTLS.  The default TLS
+// credentials use the host's root CA set.  For mTLS, pass a custom
+// grpc.DialOption built with credentials.NewTLS(tlsConfig) via a
+// private Option (or replace WithInsecure in development only).
+func New(addr string, opts ...Option) (*Client, error) {
+	c := &Client{maxRetries: 3}
+	for _, opt := range opts {
+		opt(c)
+	}
+
+	var dialOpts []grpc.DialOption
+
+	if c.insecure {
+		dialOpts = append(dialOpts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	} else {
+		// Secure TLS.  Replace by passing a custom tls.Config with
+		// ClientCAs (for mTLS) or ServerName override (for TLS).
+		dialOpts = append(dialOpts,
+			grpc.WithTransportCredentials(credentials.NewTLS(nil)))
+	}
+
+	if c.keepalive.Time > 0 || c.keepalive.Timeout > 0 {
+		dialOpts = append(dialOpts, grpc.WithKeepaliveParams(c.keepalive))
+	}
+
+	// Unary interceptor: attach API key to every RPC.
+	dialOpts = append(dialOpts, grpc.WithUnaryInterceptor(
+		func(ctx context.Context, method string, req, reply interface{},
+			cc *grpc.ClientConn, invoker grpc.UnaryInvoker, _ ...grpc.CallOption) error {
+			if c.apiKey != "" {
+				ctx = metadata.AppendToOutgoingContext(ctx,
+					"authorization", "Bearer "+c.apiKey)
+			}
+			return invoker(ctx, method, req, reply, cc)
+		}))
+
+	conn, err := grpc.NewClient(addr, dialOpts...)
+	if err != nil {
+		return nil, fmt.Errorf("grpc.Dial(%q): %w", addr, err)
+	}
+	c.conn = conn
+	c.client = pbv1.NewIngestClient(conn)
+	return c, nil
+}
+
+// Stream starts a StreamAlerts bidirectional stream.  The caller must call
+// CloseSend when done sending.  The context is used for stream deadlines.
+func (c *Client) Stream(ctx context.Context) (*Stream, error) {
+	if c.apiKey != "" {
+		md := metadata.Pairs("authorization", "Bearer "+c.apiKey)
+		ctx = metadata.NewOutgoingContext(ctx, md)
+	}
+	stream, err := c.client.StreamAlerts(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("StreamAlerts: %w", err)
+	}
+	return &Stream{stream: stream}, nil
+}
+
+// Send delivers one alert to the server.  It blocks until the server
+// acknowledges receipt (not processing — that is async).  Send retries on
+// RATE_LIMITED / UNAVAILABLE up to c.maxRetries, honouring the
+// x-retry-after-ms response header when set.
+func (s *Stream) Send(ctx context.Context, alert *pbv1.Alert) error {
+	var lastErr error
+	for attempt := 0; attempt <= 3; attempt++ {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
+
+		lastErr = s.stream.Send(alert)
+		if lastErr == nil {
+			return nil
+		}
+
+		st, ok := status.FromError(lastErr)
+		if !ok || !retryable(st.Code()) {
+			return lastErr
+		}
+
+		if attempt < 3 {
+			delay := retryDelay(s.stream.Context(), attempt)
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-time.After(delay):
+			}
+		}
+	}
+	return fmt.Errorf("Stream.Send (retries exhausted): %w", lastErr)
+}
+
+// Recv returns the next Ack from the server.  It blocks until an Ack is
+// available or the stream is closed.  It retries on UNAVAILABLE up to
+// c.maxRetries.  Returns io.EOF when the server closes the stream.
+func (s *Stream) Recv(ctx context.Context) (*pbv1.Ack, error) {
+	var lastErr error
+	for attempt := 0; attempt <= 3; attempt++ {
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		default:
+		}
+
+		ack, err := s.stream.Recv()
+		if err == nil {
+			return ack, nil
+		}
+		lastErr = err
+
+		if err == io.EOF {
+			return nil, io.EOF
+		}
+		st, ok := status.FromError(err)
+		if !ok || !retryable(st.Code()) {
+			return nil, err
+		}
+
+		if attempt < 3 {
+			delay := retryDelay(s.stream.Context(), attempt)
+			select {
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			case <-time.After(delay):
+			}
+		}
+	}
+	return nil, fmt.Errorf("Stream.Recv (retries exhausted): %w", lastErr)
+}
+
+// CloseSend signals that the caller is done sending alerts.  The server
+// will half-close its side of the stream; Recv continues to work until
+// the server closes the stream or the context is cancelled.
+func (s *Stream) CloseSend() error { return s.stream.CloseSend() }
+
+// Close tears down the underlying gRPC connection.  Call it when the
+// Client is no longer needed.
+func (c *Client) Close() error { return c.conn.Close() }
+
+// retryDelay returns the sleep interval for a given retry attempt,
+// checking the x-retry-after-ms header for a server-specified value.
+func retryDelay(streamCtx context.Context, attempt int) time.Duration {
+	if md, ok := metadata.FromIncomingContext(streamCtx); ok {
+		if vals := md.Get("x-retry-after-ms"); len(vals) > 0 {
+			var ms int
+			fmt.Sscanf(vals[0], "%d", &ms)
+			if ms > 0 {
+				return time.Duration(ms) * time.Millisecond
+			}
+		}
+	}
+	return time.Duration(100*attempt) * time.Millisecond
+}

internal/grpcclient/grpcclient_test.go

@@ -0,0 +1,266 @@
+package grpcclient
+
+import (
+	"context"
+	"io"
+	"net"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	pbv1 "git3.techno-world.net/lrosales/broad-announce/gen/go/broadannounce/v1"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/test/bufconn"
+)
+
+const bufnetSize = 1 << 20 // 1 MB
+
+// mockIngestServer implements pbv1.IngestServer for testing.
+type mockIngestServer struct {
+	pbv1.UnimplementedIngestServer
+	alertsReceived atomic.Int64
+	rateLimitAfter int64 // after N alerts, start rate-limiting
+	authKey        string
+	authFail       atomic.Bool
+}
+
+func (m *mockIngestServer) StreamAlerts(stream pbv1.Ingest_StreamAlertsServer) error {
+	md, ok := metadata.FromIncomingContext(stream.Context())
+	if !ok {
+		return io.EOF
+	}
+	if m.authFail.Load() {
+		return io.EOF
+	}
+	if len(m.authKey) > 0 {
+		vals := md.Get("authorization")
+		if len(vals) == 0 || vals[0] != "Bearer "+m.authKey {
+			return io.EOF
+		}
+	}
+
+	for {
+		alert, err := stream.Recv()
+		if err == io.EOF {
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		m.alertsReceived.Add(1)
+
+		ack := &pbv1.Ack{
+			AlertId:      alert.DedupeKey,
+			DedupeKey:    alert.DedupeKey,
+			AcceptedAtMs: time.Now().UnixMilli(),
+		}
+
+		if m.rateLimitAfter > 0 && m.alertsReceived.Load() > m.rateLimitAfter {
+			ack.Result = &pbv1.Ack_Error{
+				Error: &pbv1.Error{
+					Code:         pbv1.Error_RATE_LIMITED,
+					Message:      "rate limited by test server",
+					RetryAfterMs: 10,
+				},
+			}
+		} else {
+			ack.Result = &pbv1.Ack_Ok{
+				Ok: &pbv1.Ok{},
+			}
+		}
+
+		if err := stream.Send(ack); err != nil {
+			return err
+		}
+	}
+}
+
+func newTestServer(t *testing.T) (*grpc.Server, *bufconn.Listener) {
+	lis := bufconn.Listen(bufnetSize)
+	srv := grpc.NewServer()
+	pbv1.RegisterIngestServer(srv, &mockIngestServer{authKey: "acme-001:prom:s3cret"})
+	go srv.Serve(lis)
+	return srv, lis
+}
+
+func dialBufconn(ctx context.Context, t *testing.T, lis *bufconn.Listener) *grpc.ClientConn {
+	conn, err := grpc.NewClient(
+		"passthrough://bufconn",
+		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
+			return lis.Dial()
+		}),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	if err != nil {
+		t.Fatalf("grpc.NewClient: %v", err)
+	}
+	return conn
+}
+
+func TestClient_New(t *testing.T) {
+	c, err := New("localhost:9090",
+		WithAPIKey("acme-001:prom:s3cret"),
+		WithMaxRetries(5),
+		WithInsecure(),
+	)
+	if err != nil {
+		t.Fatalf("New() error = %v", err)
+	}
+	if c == nil {
+		t.Fatal("New() returned nil client")
+	}
+	if c.maxRetries != 5 {
+		t.Errorf("maxRetries = %d, want 5", c.maxRetries)
+	}
+	c.Close()
+}
+
+func TestClient_OptionDefaults(t *testing.T) {
+	c, err := New("localhost:9090", WithInsecure())
+	if err != nil {
+		t.Fatalf("New() error = %v", err)
+	}
+	if c.maxRetries != 3 {
+		t.Errorf("default maxRetries = %d, want 3", c.maxRetries)
+	}
+	c.Close()
+}
+
+func TestStream_SendRecv(t *testing.T) {
+	ctx := context.Background()
+	srv, lis := newTestServer(t)
+	defer srv.Stop()
+
+	conn := dialBufconn(ctx, t, lis)
+	defer conn.Close()
+
+	client := pbv1.NewIngestClient(conn)
+	md := metadata.Pairs("authorization", "Bearer acme-001:prom:s3cret")
+	sctx := metadata.NewOutgoingContext(ctx, md)
+
+	stream, err := client.StreamAlerts(sctx)
+	if err != nil {
+		t.Fatalf("StreamAlerts() error = %v", err)
+	}
+
+	alert := &pbv1.Alert{
+		CompanyId:  "acme-001",
+		SourceId:   "prom",
+		Severity:   "critical",
+		Title:      "test alert",
+		DedupeKey:  "dk-001",
+		ClientTsMs: time.Now().UnixMilli(),
+	}
+	if err := stream.Send(alert); err != nil {
+		t.Fatalf("stream.Send() error = %v", err)
+	}
+
+	ack, err := stream.Recv()
+	if err != nil {
+		t.Fatalf("stream.Recv() error = %v", err)
+	}
+	if ack.DedupeKey != "dk-001" {
+		t.Errorf("ack.DedupeKey = %q, want %q", ack.DedupeKey, "dk-001")
+	}
+
+	stream.CloseSend()
+}
+
+func TestStream_CloseSend(t *testing.T) {
+	ctx := context.Background()
+	srv, lis := newTestServer(t)
+	defer srv.Stop()
+
+	conn := dialBufconn(ctx, t, lis)
+	defer conn.Close()
+
+	client := pbv1.NewIngestClient(conn)
+	md := metadata.Pairs("authorization", "Bearer acme-001:prom:s3cret")
+	sctx := metadata.NewOutgoingContext(ctx, md)
+
+	stream, err := client.StreamAlerts(sctx)
+	if err != nil {
+		t.Fatalf("StreamAlerts() error = %v", err)
+	}
+
+	if err := stream.CloseSend(); err != nil {
+		t.Errorf("CloseSend() error = %v", err)
+	}
+
+	_, err = stream.Recv()
+	if err != io.EOF {
+		t.Errorf("Recv() after CloseSend = %v, want io.EOF", err)
+	}
+}
+
+func TestRetryable(t *testing.T) {
+	tests := []struct {
+		code codes.Code
+		want bool
+	}{
+		{codes.Unavailable, true},
+		{codes.ResourceExhausted, true},
+		{codes.Internal, true},
+		{codes.OK, false},
+		{codes.InvalidArgument, false},
+		{codes.NotFound, false},
+		{codes.Unauthenticated, false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.code.String(), func(t *testing.T) {
+			if got := retryable(tt.code); got != tt.want {
+				t.Errorf("retryable(%v) = %v, want %v", tt.code, got, tt.want)
+			}
+		})
+	}
+}
+
+func TestStream_SendRecv_MultipleAlerts(t *testing.T) {
+	ctx := context.Background()
+	srv, lis := newTestServer(t)
+	defer srv.Stop()
+
+	conn := dialBufconn(ctx, t, lis)
+	defer conn.Close()
+
+	client := pbv1.NewIngestClient(conn)
+	md := metadata.Pairs("authorization", "Bearer acme-001:prom:s3cret")
+	sctx := metadata.NewOutgoingContext(ctx, md)
+
+	stream, err := client.StreamAlerts(sctx)
+	if err != nil {
+		t.Fatalf("StreamAlerts() error = %v", err)
+	}
+
+	const n = 5
+	for i := 0; i < n; i++ {
+		alert := &pbv1.Alert{
+			CompanyId:  "acme-001",
+			SourceId:   "prom",
+			Severity:   "info",
+			Title:      "test alert",
+			DedupeKey:  "dk-multi-" + string(rune('0'+i)),
+			ClientTsMs: time.Now().UnixMilli(),
+		}
+		if err := stream.Send(alert); err != nil {
+			t.Fatalf("stream.Send() error = %v", err)
+		}
+		ack, err := stream.Recv()
+		if err != nil {
+			t.Fatalf("stream.Recv() error = %v", err)
+		}
+		if ack.DedupeKey != alert.DedupeKey {
+			t.Errorf("ack.DedupeKey = %q, want %q", ack.DedupeKey, alert.DedupeKey)
+		}
+	}
+	stream.CloseSend()
+}
+
+// Verify generated types implement the expected interfaces.
+func TestProtoInterfaces(t *testing.T) {
+	var _ pbv1.IngestClient = nil
+	var _ pbv1.IngestServer = nil
+}

internal/grpcclient/options.go

@@ -0,0 +1,48 @@
+package grpcclient
+
+import (
+	"time"
+
+	"google.golang.org/grpc/keepalive"
+)
+
+// Option configures a Client.
+type Option func(*Client)
+
+// WithAPIKey sets the API key used for every StreamAlerts stream.
+// The key is sent as "Bearer <key>" in the grpc-metadata
+// "authorization" header on each stream.
+//
+// The format is "<company_id>:<source_id>:<secret>".
+func WithAPIKey(k string) Option {
+	return func(c *Client) { c.apiKey = k }
+}
+
+// WithMaxRetries sets the maximum number of retry attempts for transient
+// errors (RATE_LIMITED, UNAVAILABLE).  Zero disables retries.
+// The backoff between retries follows Error.retry_after_ms from the server
+// when available, otherwise exponential backoff starting at 100ms.
+func WithMaxRetries(n int) Option {
+	return func(c *Client) { c.maxRetries = n }
+}
+
+// WithKeepalive configures the gRPC keepalive parameters on the underlying
+// connection. Time is the interval between keepalive probes; Timeout is how
+// long the peer has to respond before the connection is closed.
+func WithKeepalive(time, timeout time.Duration) Option {
+	return func(c *Client) {
+		c.keepalive = keepalive.ClientParameters{
+			Time:                time,
+			Timeout:             timeout,
+			PermitWithoutStream: false,
+		}
+	}
+}
+
+// WithInsecure disables transport security.  Appropriate for development
+// only (e.g. connecting to ingestd on localhost).  In production, connections
+// to ingestd should use mTLS or at minimum TLS, configured via a
+// grpc.DialOption built with credentials.NewTLS.
+func WithInsecure() Option {
+	return func(c *Client) { c.insecure = true }
+}