feat: commit generated Go code, consume via go get instead of submodule

Add go.mod, buf.gen.yaml, and generated gen/kafkamgmt/v1 so consumers
fetch this module directly via go get against the self-hosted Forgejo
host, with no local buf/protoc plugin installation required. Add a
codegen-check CI job to keep gen/ in sync with proto/.

.forgejo/workflows/ci.yaml

@@ -20,3 +20,26 @@ jobs:
       - name: buf breaking (against main)
         run: buf breaking --against '.git#branch=main'
         if: github.ref != 'refs/heads/main'
+
+  codegen-check:
+    runs-on: docker
+    container:
+      image: golang:1.25
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Install buf and protoc plugins
+        run: |
+          go install github.com/bufbuild/buf/cmd/buf@latest
+          go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
+          go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest
+          go install github.com/grpc-ecosystem/grpc-gateway/v2/protoc-gen-grpc-gateway@latest
+
+      - name: Regenerate and diff
+        run: |
+          export PATH="$PATH:$(go env GOPATH)/bin"
+          buf generate
+          git diff --exit-code -- gen || (echo "::error::gen/ is out of date — run buf generate and commit the result" && exit 1)
+
+      - name: go build
+        run: go build ./...

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 All notable wire-contract changes to `kafkamgmt.v1` are recorded here, one entry per tag. Follows semantic versioning: **minor** for additive/compatible changes, **major** for anything `buf breaking` would flag.
 
-## v1.0.0 (unreleased)
+## v1.1.0
 
-Initial extraction from `kafaka_management_service` (`proto/kafkamgmt/v1/queue_service.proto`), originating commit `1373b9e`. `QueueService` with 6 data-plane RPCs: `SendMessage`, `SendMessageBatch`, `ReceiveMessage`, `DeleteMessage`, `DeleteMessageBatch`, `ChangeMessageVisibility`. No wire-format changes from the source — only `go_package` was changed to a neutral placeholder (see `proto/kafkamgmt/v1/queue_service.proto` header comment); this has no effect on the wire format.
+Switched to committed-generated-code model: `go.mod` added (module `forgejo.riotpiao.homelab.com/rock/kmsvc-proto`), `go_package` repointed at this module's own path, generated Go code committed under `gen/kafkamgmt/v1`. Consumers now `go get` this module directly instead of submoduling + running their own `buf generate`. No wire-format changes.
+
+## v1.0.0
+
+Initial extraction from `kafaka_management_service` (`proto/kafkamgmt/v1/queue_service.proto`), originating commit `1373b9e`. `QueueService` with 6 data-plane RPCs: `SendMessage`, `SendMessageBatch`, `ReceiveMessage`, `DeleteMessage`, `DeleteMessageBatch`, `ChangeMessageVisibility`. No wire-format changes from the source — only `go_package` was changed to a neutral placeholder.

README.md

@@ -1,32 +1,30 @@
 # kmsvc-proto
 
-Shared `kafkamgmt.v1` proto contract for the Kafka Management Service ecosystem (`kafaka_management_service`, `kmsvc-sdk`, `kmsvc-cli`).
+Shared `kafkamgmt.v1` proto contract and generated Go code for the Kafka Management Service ecosystem (`kafaka_management_service`, `kmsvc-sdk`, `kmsvc-cli`).
 
-This repo ships **`.proto` source + buf lint/breaking-change config only** — no generated code. Each consumer submodules this repo and runs its own `buf generate` with its own pinned plugin versions and `go_package` override (managed mode), so generated-runtime versions never need to match across repos — only the wire contract is shared.
+This repo ships **`.proto` source + pre-generated, committed Go code** (`gen/kafkamgmt/v1`). Generation happens once here, in this repo, not in every consumer. Consumers never install `buf`/`protoc` plugins — they just `go get` this module like any other Go dependency.
 
 ## Consuming this repo
 
 ```bash
-git submodule add https://forgejo.riotpiao.homelab.com/rock/kmsvc-proto.git proto
+export GOPRIVATE=forgejo.riotpiao.homelab.com   # skip public proxy/sumdb for this host
-git -C proto checkout v1.0.0   # pin to a tag, don't track a moving branch
+go get forgejo.riotpiao.homelab.com/rock/[email protected]
 ```
 
-Then point your own `buf.gen.yaml` at `proto/proto` (this repo's proto root) and use managed mode to set your module's own `go_package`:
+```go
-
+import kafkamgmtv1 "forgejo.riotpiao.homelab.com/rock/kmsvc-proto/gen/kafkamgmt/v1"
-```yaml
-version: v2
-managed:
-  enabled: true
-  override:
-    - file_option: go_package
-      value: github.com/<you>/<your-module>/internal/genapi
-plugins:
-  - local: protoc-gen-go
-    out: internal/genapi
-    opt: paths=source_relative
-  - local: protoc-gen-go-grpc
-    out: internal/genapi
-    opt: paths=source_relative
 ```
 
-See `PLAN.md` for the full rollout plan (tagging convention, server migration, SDK/CLI wiring).
+## Updating the contract
+
+```bash
+# edit proto/kafkamgmt/v1/queue_service.proto
+buf lint
+buf generate   # regenerates gen/ in place
+go build ./...
+git add -A && git commit -m "..." && git tag vX.Y.Z && git push --tags
+```
+
+CI (`buf-lint-and-breaking`) gates lint + breaking-change checks. CI (`codegen-check`) re-runs `buf generate` and fails if `gen/` doesn't match what's committed — never hand-edit `gen/`.
+
+See `PLAN.md` for the full rollout plan (tagging convention, server/SDK/CLI migration).

buf.gen.yaml

@@ -0,0 +1,11 @@
+version: v2
+plugins:
+  - local: protoc-gen-go
+    out: gen
+    opt: paths=source_relative
+  - local: protoc-gen-go-grpc
+    out: gen
+    opt: paths=source_relative
+  - local: protoc-gen-grpc-gateway
+    out: gen
+    opt: paths=source_relative

gen/kafkamgmt/v1/queue_service.pb.gw.go

@@ -0,0 +1,629 @@
+// Code generated by protoc-gen-grpc-gateway. DO NOT EDIT.
+// source: kafkamgmt/v1/queue_service.proto
+
+/*
+Package kafkamgmtv1 is a reverse proxy.
+
+It translates gRPC into RESTful JSON APIs.
+*/
+package kafkamgmtv1
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net/http"
+
+	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
+	"github.com/grpc-ecosystem/grpc-gateway/v2/utilities"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/grpclog"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
+)
+
+// Suppress "imported and not used" errors
+var (
+	_ codes.Code
+	_ io.Reader
+	_ status.Status
+	_ = errors.New
+	_ = runtime.String
+	_ = utilities.NewDoubleArray
+	_ = metadata.Join
+)
+
+func request_QueueService_SendMessage_0(ctx context.Context, marshaler runtime.Marshaler, client QueueServiceClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+	var (
+		protoReq SendMessageRequest
+		metadata runtime.ServerMetadata
+		err      error
+	)
+	if err := marshaler.NewDecoder(req.Body).Decode(&protoReq); err != nil && !errors.Is(err, io.EOF) {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+	}
+	if req.Body != nil {
+		_, _ = io.Copy(io.Discard, req.Body)
+	}
+	val, ok := pathParams["queue_name"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "queue_name")
+	}
+	protoReq.QueueName, err = runtime.String(val)
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "queue_name", err)
+	}
+	msg, err := client.SendMessage(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
+	return msg, metadata, err
+}
+
+func local_request_QueueService_SendMessage_0(ctx context.Context, marshaler runtime.Marshaler, server QueueServiceServer, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+	var (
+		protoReq SendMessageRequest
+		metadata runtime.ServerMetadata
+		err      error
+	)
+	if err := marshaler.NewDecoder(req.Body).Decode(&protoReq); err != nil && !errors.Is(err, io.EOF) {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+	}
+	val, ok := pathParams["queue_name"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "queue_name")
+	}
+	protoReq.QueueName, err = runtime.String(val)
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "queue_name", err)
+	}
+	msg, err := server.SendMessage(ctx, &protoReq)
+	return msg, metadata, err
+}
+
+func request_QueueService_SendMessageBatch_0(ctx context.Context, marshaler runtime.Marshaler, client QueueServiceClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+	var (
+		protoReq SendMessageBatchRequest
+		metadata runtime.ServerMetadata
+		err      error
+	)
+	if err := marshaler.NewDecoder(req.Body).Decode(&protoReq); err != nil && !errors.Is(err, io.EOF) {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+	}
+	if req.Body != nil {
+		_, _ = io.Copy(io.Discard, req.Body)
+	}
+	val, ok := pathParams["queue_name"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "queue_name")
+	}
+	protoReq.QueueName, err = runtime.String(val)
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "queue_name", err)
+	}
+	msg, err := client.SendMessageBatch(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
+	return msg, metadata, err
+}
+
+func local_request_QueueService_SendMessageBatch_0(ctx context.Context, marshaler runtime.Marshaler, server QueueServiceServer, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+	var (
+		protoReq SendMessageBatchRequest
+		metadata runtime.ServerMetadata
+		err      error
+	)
+	if err := marshaler.NewDecoder(req.Body).Decode(&protoReq); err != nil && !errors.Is(err, io.EOF) {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+	}
+	val, ok := pathParams["queue_name"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "queue_name")
+	}
+	protoReq.QueueName, err = runtime.String(val)
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "queue_name", err)
+	}
+	msg, err := server.SendMessageBatch(ctx, &protoReq)
+	return msg, metadata, err
+}
+
+var filter_QueueService_ReceiveMessage_0 = &utilities.DoubleArray{Encoding: map[string]int{"queue_name": 0}, Base: []int{1, 1, 0}, Check: []int{0, 1, 2}}
+
+func request_QueueService_ReceiveMessage_0(ctx context.Context, marshaler runtime.Marshaler, client QueueServiceClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+	var (
+		protoReq ReceiveMessageRequest
+		metadata runtime.ServerMetadata
+		err      error
+	)
+	if req.Body != nil {
+		_, _ = io.Copy(io.Discard, req.Body)
+	}
+	val, ok := pathParams["queue_name"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "queue_name")
+	}
+	protoReq.QueueName, err = runtime.String(val)
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "queue_name", err)
+	}
+	if err := req.ParseForm(); err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+	}
+	if err := runtime.PopulateQueryParameters(&protoReq, req.Form, filter_QueueService_ReceiveMessage_0); err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+	}
+	msg, err := client.ReceiveMessage(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
+	return msg, metadata, err
+}
+
+func local_request_QueueService_ReceiveMessage_0(ctx context.Context, marshaler runtime.Marshaler, server QueueServiceServer, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+	var (
+		protoReq ReceiveMessageRequest
+		metadata runtime.ServerMetadata
+		err      error
+	)
+	val, ok := pathParams["queue_name"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "queue_name")
+	}
+	protoReq.QueueName, err = runtime.String(val)
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "queue_name", err)
+	}
+	if err := req.ParseForm(); err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+	}
+	if err := runtime.PopulateQueryParameters(&protoReq, req.Form, filter_QueueService_ReceiveMessage_0); err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+	}
+	msg, err := server.ReceiveMessage(ctx, &protoReq)
+	return msg, metadata, err
+}
+
+func request_QueueService_DeleteMessage_0(ctx context.Context, marshaler runtime.Marshaler, client QueueServiceClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+	var (
+		protoReq DeleteMessageRequest
+		metadata runtime.ServerMetadata
+		err      error
+	)
+	if req.Body != nil {
+		_, _ = io.Copy(io.Discard, req.Body)
+	}
+	val, ok := pathParams["queue_name"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "queue_name")
+	}
+	protoReq.QueueName, err = runtime.String(val)
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "queue_name", err)
+	}
+	val, ok = pathParams["receipt_handle"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "receipt_handle")
+	}
+	protoReq.ReceiptHandle, err = runtime.String(val)
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "receipt_handle", err)
+	}
+	msg, err := client.DeleteMessage(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
+	return msg, metadata, err
+}
+
+func local_request_QueueService_DeleteMessage_0(ctx context.Context, marshaler runtime.Marshaler, server QueueServiceServer, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+	var (
+		protoReq DeleteMessageRequest
+		metadata runtime.ServerMetadata
+		err      error
+	)
+	val, ok := pathParams["queue_name"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "queue_name")
+	}
+	protoReq.QueueName, err = runtime.String(val)
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "queue_name", err)
+	}
+	val, ok = pathParams["receipt_handle"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "receipt_handle")
+	}
+	protoReq.ReceiptHandle, err = runtime.String(val)
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "receipt_handle", err)
+	}
+	msg, err := server.DeleteMessage(ctx, &protoReq)
+	return msg, metadata, err
+}
+
+func request_QueueService_DeleteMessageBatch_0(ctx context.Context, marshaler runtime.Marshaler, client QueueServiceClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+	var (
+		protoReq DeleteMessageBatchRequest
+		metadata runtime.ServerMetadata
+		err      error
+	)
+	if err := marshaler.NewDecoder(req.Body).Decode(&protoReq); err != nil && !errors.Is(err, io.EOF) {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+	}
+	if req.Body != nil {
+		_, _ = io.Copy(io.Discard, req.Body)
+	}
+	val, ok := pathParams["queue_name"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "queue_name")
+	}
+	protoReq.QueueName, err = runtime.String(val)
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "queue_name", err)
+	}
+	msg, err := client.DeleteMessageBatch(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
+	return msg, metadata, err
+}
+
+func local_request_QueueService_DeleteMessageBatch_0(ctx context.Context, marshaler runtime.Marshaler, server QueueServiceServer, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+	var (
+		protoReq DeleteMessageBatchRequest
+		metadata runtime.ServerMetadata
+		err      error
+	)
+	if err := marshaler.NewDecoder(req.Body).Decode(&protoReq); err != nil && !errors.Is(err, io.EOF) {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+	}
+	val, ok := pathParams["queue_name"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "queue_name")
+	}
+	protoReq.QueueName, err = runtime.String(val)
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "queue_name", err)
+	}
+	msg, err := server.DeleteMessageBatch(ctx, &protoReq)
+	return msg, metadata, err
+}
+
+func request_QueueService_ChangeMessageVisibility_0(ctx context.Context, marshaler runtime.Marshaler, client QueueServiceClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+	var (
+		protoReq ChangeMessageVisibilityRequest
+		metadata runtime.ServerMetadata
+		err      error
+	)
+	if err := marshaler.NewDecoder(req.Body).Decode(&protoReq); err != nil && !errors.Is(err, io.EOF) {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+	}
+	if req.Body != nil {
+		_, _ = io.Copy(io.Discard, req.Body)
+	}
+	val, ok := pathParams["queue_name"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "queue_name")
+	}
+	protoReq.QueueName, err = runtime.String(val)
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "queue_name", err)
+	}
+	val, ok = pathParams["receipt_handle"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "receipt_handle")
+	}
+	protoReq.ReceiptHandle, err = runtime.String(val)
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "receipt_handle", err)
+	}
+	msg, err := client.ChangeMessageVisibility(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
+	return msg, metadata, err
+}
+
+func local_request_QueueService_ChangeMessageVisibility_0(ctx context.Context, marshaler runtime.Marshaler, server QueueServiceServer, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+	var (
+		protoReq ChangeMessageVisibilityRequest
+		metadata runtime.ServerMetadata
+		err      error
+	)
+	if err := marshaler.NewDecoder(req.Body).Decode(&protoReq); err != nil && !errors.Is(err, io.EOF) {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+	}
+	val, ok := pathParams["queue_name"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "queue_name")
+	}
+	protoReq.QueueName, err = runtime.String(val)
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "queue_name", err)
+	}
+	val, ok = pathParams["receipt_handle"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "receipt_handle")
+	}
+	protoReq.ReceiptHandle, err = runtime.String(val)
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "receipt_handle", err)
+	}
+	msg, err := server.ChangeMessageVisibility(ctx, &protoReq)
+	return msg, metadata, err
+}
+
+// RegisterQueueServiceHandlerServer registers the http handlers for service QueueService to "mux".
+// UnaryRPC     :call QueueServiceServer directly.
+// StreamingRPC :currently unsupported pending https://github.com/grpc/grpc-go/issues/906.
+// Note that using this registration option will cause many gRPC library features to stop working. Consider using RegisterQueueServiceHandlerFromEndpoint instead.
+// GRPC interceptors will not work for this type of registration. To use interceptors, you must use the "runtime.WithMiddlewares" option in the "runtime.NewServeMux" call.
+func RegisterQueueServiceHandlerServer(ctx context.Context, mux *runtime.ServeMux, server QueueServiceServer) error {
+	mux.Handle(http.MethodPost, pattern_QueueService_SendMessage_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		var stream runtime.ServerTransportStream
+		ctx = grpc.NewContextWithServerTransportStream(ctx, &stream)
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		annotatedContext, err := runtime.AnnotateIncomingContext(ctx, mux, req, "/kafkamgmt.v1.QueueService/SendMessage", runtime.WithHTTPPathPattern("/v1/queues/{queue_name}/messages"))
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := local_request_QueueService_SendMessage_0(annotatedContext, inboundMarshaler, server, req, pathParams)
+		md.HeaderMD, md.TrailerMD = metadata.Join(md.HeaderMD, stream.Header()), metadata.Join(md.TrailerMD, stream.Trailer())
+		annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md)
+		if err != nil {
+			runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		forward_QueueService_SendMessage_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+	})
+	mux.Handle(http.MethodPost, pattern_QueueService_SendMessageBatch_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		var stream runtime.ServerTransportStream
+		ctx = grpc.NewContextWithServerTransportStream(ctx, &stream)
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		annotatedContext, err := runtime.AnnotateIncomingContext(ctx, mux, req, "/kafkamgmt.v1.QueueService/SendMessageBatch", runtime.WithHTTPPathPattern("/v1/queues/{queue_name}/messages:batch"))
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := local_request_QueueService_SendMessageBatch_0(annotatedContext, inboundMarshaler, server, req, pathParams)
+		md.HeaderMD, md.TrailerMD = metadata.Join(md.HeaderMD, stream.Header()), metadata.Join(md.TrailerMD, stream.Trailer())
+		annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md)
+		if err != nil {
+			runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		forward_QueueService_SendMessageBatch_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+	})
+	mux.Handle(http.MethodGet, pattern_QueueService_ReceiveMessage_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		var stream runtime.ServerTransportStream
+		ctx = grpc.NewContextWithServerTransportStream(ctx, &stream)
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		annotatedContext, err := runtime.AnnotateIncomingContext(ctx, mux, req, "/kafkamgmt.v1.QueueService/ReceiveMessage", runtime.WithHTTPPathPattern("/v1/queues/{queue_name}/messages"))
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := local_request_QueueService_ReceiveMessage_0(annotatedContext, inboundMarshaler, server, req, pathParams)
+		md.HeaderMD, md.TrailerMD = metadata.Join(md.HeaderMD, stream.Header()), metadata.Join(md.TrailerMD, stream.Trailer())
+		annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md)
+		if err != nil {
+			runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		forward_QueueService_ReceiveMessage_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+	})
+	mux.Handle(http.MethodDelete, pattern_QueueService_DeleteMessage_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		var stream runtime.ServerTransportStream
+		ctx = grpc.NewContextWithServerTransportStream(ctx, &stream)
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		annotatedContext, err := runtime.AnnotateIncomingContext(ctx, mux, req, "/kafkamgmt.v1.QueueService/DeleteMessage", runtime.WithHTTPPathPattern("/v1/queues/{queue_name}/messages/{receipt_handle}"))
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := local_request_QueueService_DeleteMessage_0(annotatedContext, inboundMarshaler, server, req, pathParams)
+		md.HeaderMD, md.TrailerMD = metadata.Join(md.HeaderMD, stream.Header()), metadata.Join(md.TrailerMD, stream.Trailer())
+		annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md)
+		if err != nil {
+			runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		forward_QueueService_DeleteMessage_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+	})
+	mux.Handle(http.MethodPost, pattern_QueueService_DeleteMessageBatch_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		var stream runtime.ServerTransportStream
+		ctx = grpc.NewContextWithServerTransportStream(ctx, &stream)
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		annotatedContext, err := runtime.AnnotateIncomingContext(ctx, mux, req, "/kafkamgmt.v1.QueueService/DeleteMessageBatch", runtime.WithHTTPPathPattern("/v1/queues/{queue_name}/messages:batchDelete"))
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := local_request_QueueService_DeleteMessageBatch_0(annotatedContext, inboundMarshaler, server, req, pathParams)
+		md.HeaderMD, md.TrailerMD = metadata.Join(md.HeaderMD, stream.Header()), metadata.Join(md.TrailerMD, stream.Trailer())
+		annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md)
+		if err != nil {
+			runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		forward_QueueService_DeleteMessageBatch_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+	})
+	mux.Handle(http.MethodPatch, pattern_QueueService_ChangeMessageVisibility_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		var stream runtime.ServerTransportStream
+		ctx = grpc.NewContextWithServerTransportStream(ctx, &stream)
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		annotatedContext, err := runtime.AnnotateIncomingContext(ctx, mux, req, "/kafkamgmt.v1.QueueService/ChangeMessageVisibility", runtime.WithHTTPPathPattern("/v1/queues/{queue_name}/messages/{receipt_handle}"))
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := local_request_QueueService_ChangeMessageVisibility_0(annotatedContext, inboundMarshaler, server, req, pathParams)
+		md.HeaderMD, md.TrailerMD = metadata.Join(md.HeaderMD, stream.Header()), metadata.Join(md.TrailerMD, stream.Trailer())
+		annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md)
+		if err != nil {
+			runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		forward_QueueService_ChangeMessageVisibility_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+	})
+
+	return nil
+}
+
+// RegisterQueueServiceHandlerFromEndpoint is same as RegisterQueueServiceHandler but
+// automatically dials to "endpoint" and closes the connection when "ctx" gets done.
+func RegisterQueueServiceHandlerFromEndpoint(ctx context.Context, mux *runtime.ServeMux, endpoint string, opts []grpc.DialOption) (err error) {
+	conn, err := grpc.NewClient(endpoint, opts...)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if err != nil {
+			if cerr := conn.Close(); cerr != nil {
+				grpclog.Errorf("Failed to close conn to %s: %v", endpoint, cerr)
+			}
+			return
+		}
+		go func() {
+			<-ctx.Done()
+			if cerr := conn.Close(); cerr != nil {
+				grpclog.Errorf("Failed to close conn to %s: %v", endpoint, cerr)
+			}
+		}()
+	}()
+	return RegisterQueueServiceHandler(ctx, mux, conn)
+}
+
+// RegisterQueueServiceHandler registers the http handlers for service QueueService to "mux".
+// The handlers forward requests to the grpc endpoint over "conn".
+func RegisterQueueServiceHandler(ctx context.Context, mux *runtime.ServeMux, conn *grpc.ClientConn) error {
+	return RegisterQueueServiceHandlerClient(ctx, mux, NewQueueServiceClient(conn))
+}
+
+// RegisterQueueServiceHandlerClient registers the http handlers for service QueueService
+// to "mux". The handlers forward requests to the grpc endpoint over the given implementation of "QueueServiceClient".
+// Note: the gRPC framework executes interceptors within the gRPC handler. If the passed in "QueueServiceClient"
+// doesn't go through the normal gRPC flow (creating a gRPC client etc.) then it will be up to the passed in
+// "QueueServiceClient" to call the correct interceptors. This client ignores the HTTP middlewares.
+func RegisterQueueServiceHandlerClient(ctx context.Context, mux *runtime.ServeMux, client QueueServiceClient) error {
+	mux.Handle(http.MethodPost, pattern_QueueService_SendMessage_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		annotatedContext, err := runtime.AnnotateContext(ctx, mux, req, "/kafkamgmt.v1.QueueService/SendMessage", runtime.WithHTTPPathPattern("/v1/queues/{queue_name}/messages"))
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := request_QueueService_SendMessage_0(annotatedContext, inboundMarshaler, client, req, pathParams)
+		annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md)
+		if err != nil {
+			runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		forward_QueueService_SendMessage_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+	})
+	mux.Handle(http.MethodPost, pattern_QueueService_SendMessageBatch_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		annotatedContext, err := runtime.AnnotateContext(ctx, mux, req, "/kafkamgmt.v1.QueueService/SendMessageBatch", runtime.WithHTTPPathPattern("/v1/queues/{queue_name}/messages:batch"))
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := request_QueueService_SendMessageBatch_0(annotatedContext, inboundMarshaler, client, req, pathParams)
+		annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md)
+		if err != nil {
+			runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		forward_QueueService_SendMessageBatch_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+	})
+	mux.Handle(http.MethodGet, pattern_QueueService_ReceiveMessage_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		annotatedContext, err := runtime.AnnotateContext(ctx, mux, req, "/kafkamgmt.v1.QueueService/ReceiveMessage", runtime.WithHTTPPathPattern("/v1/queues/{queue_name}/messages"))
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := request_QueueService_ReceiveMessage_0(annotatedContext, inboundMarshaler, client, req, pathParams)
+		annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md)
+		if err != nil {
+			runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		forward_QueueService_ReceiveMessage_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+	})
+	mux.Handle(http.MethodDelete, pattern_QueueService_DeleteMessage_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		annotatedContext, err := runtime.AnnotateContext(ctx, mux, req, "/kafkamgmt.v1.QueueService/DeleteMessage", runtime.WithHTTPPathPattern("/v1/queues/{queue_name}/messages/{receipt_handle}"))
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := request_QueueService_DeleteMessage_0(annotatedContext, inboundMarshaler, client, req, pathParams)
+		annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md)
+		if err != nil {
+			runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		forward_QueueService_DeleteMessage_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+	})
+	mux.Handle(http.MethodPost, pattern_QueueService_DeleteMessageBatch_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		annotatedContext, err := runtime.AnnotateContext(ctx, mux, req, "/kafkamgmt.v1.QueueService/DeleteMessageBatch", runtime.WithHTTPPathPattern("/v1/queues/{queue_name}/messages:batchDelete"))
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := request_QueueService_DeleteMessageBatch_0(annotatedContext, inboundMarshaler, client, req, pathParams)
+		annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md)
+		if err != nil {
+			runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		forward_QueueService_DeleteMessageBatch_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+	})
+	mux.Handle(http.MethodPatch, pattern_QueueService_ChangeMessageVisibility_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		annotatedContext, err := runtime.AnnotateContext(ctx, mux, req, "/kafkamgmt.v1.QueueService/ChangeMessageVisibility", runtime.WithHTTPPathPattern("/v1/queues/{queue_name}/messages/{receipt_handle}"))
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := request_QueueService_ChangeMessageVisibility_0(annotatedContext, inboundMarshaler, client, req, pathParams)
+		annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md)
+		if err != nil {
+			runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		forward_QueueService_ChangeMessageVisibility_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+	})
+	return nil
+}
+
+var (
+	pattern_QueueService_SendMessage_0             = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 1, 0, 4, 1, 5, 2, 2, 3}, []string{"v1", "queues", "queue_name", "messages"}, ""))
+	pattern_QueueService_SendMessageBatch_0        = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 1, 0, 4, 1, 5, 2, 2, 3}, []string{"v1", "queues", "queue_name", "messages"}, "batch"))
+	pattern_QueueService_ReceiveMessage_0          = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 1, 0, 4, 1, 5, 2, 2, 3}, []string{"v1", "queues", "queue_name", "messages"}, ""))
+	pattern_QueueService_DeleteMessage_0           = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 1, 0, 4, 1, 5, 2, 2, 3, 1, 0, 4, 1, 5, 4}, []string{"v1", "queues", "queue_name", "messages", "receipt_handle"}, ""))
+	pattern_QueueService_DeleteMessageBatch_0      = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 1, 0, 4, 1, 5, 2, 2, 3}, []string{"v1", "queues", "queue_name", "messages"}, "batchDelete"))
+	pattern_QueueService_ChangeMessageVisibility_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 1, 0, 4, 1, 5, 2, 2, 3, 1, 0, 4, 1, 5, 4}, []string{"v1", "queues", "queue_name", "messages", "receipt_handle"}, ""))
+)
+
+var (
+	forward_QueueService_SendMessage_0             = runtime.ForwardResponseMessage
+	forward_QueueService_SendMessageBatch_0        = runtime.ForwardResponseMessage
+	forward_QueueService_ReceiveMessage_0          = runtime.ForwardResponseMessage
+	forward_QueueService_DeleteMessage_0           = runtime.ForwardResponseMessage
+	forward_QueueService_DeleteMessageBatch_0      = runtime.ForwardResponseMessage
+	forward_QueueService_ChangeMessageVisibility_0 = runtime.ForwardResponseMessage
+)

gen/kafkamgmt/v1/queue_service_grpc.pb.go

@@ -0,0 +1,319 @@
+// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.6.2
+// - protoc             (unknown)
+// source: kafkamgmt/v1/queue_service.proto
+
+package kafkamgmtv1
+
+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 (
+	QueueService_SendMessage_FullMethodName             = "/kafkamgmt.v1.QueueService/SendMessage"
+	QueueService_SendMessageBatch_FullMethodName        = "/kafkamgmt.v1.QueueService/SendMessageBatch"
+	QueueService_ReceiveMessage_FullMethodName          = "/kafkamgmt.v1.QueueService/ReceiveMessage"
+	QueueService_DeleteMessage_FullMethodName           = "/kafkamgmt.v1.QueueService/DeleteMessage"
+	QueueService_DeleteMessageBatch_FullMethodName      = "/kafkamgmt.v1.QueueService/DeleteMessageBatch"
+	QueueService_ChangeMessageVisibility_FullMethodName = "/kafkamgmt.v1.QueueService/ChangeMessageVisibility"
+)
+
+// QueueServiceClient is the client API for QueueService 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.
+//
+// QueueService is the message-plane API for the Kafka Management Service.
+// Queue lifecycle (create/delete/configure) is managed via the Queue CRD,
+// not this service — see design.md §2a/§2b in kafaka_management_service.
+type QueueServiceClient interface {
+	SendMessage(ctx context.Context, in *SendMessageRequest, opts ...grpc.CallOption) (*SendMessageResponse, error)
+	SendMessageBatch(ctx context.Context, in *SendMessageBatchRequest, opts ...grpc.CallOption) (*SendMessageBatchResponse, error)
+	ReceiveMessage(ctx context.Context, in *ReceiveMessageRequest, opts ...grpc.CallOption) (*ReceiveMessageResponse, error)
+	DeleteMessage(ctx context.Context, in *DeleteMessageRequest, opts ...grpc.CallOption) (*DeleteMessageResponse, error)
+	DeleteMessageBatch(ctx context.Context, in *DeleteMessageBatchRequest, opts ...grpc.CallOption) (*DeleteMessageBatchResponse, error)
+	ChangeMessageVisibility(ctx context.Context, in *ChangeMessageVisibilityRequest, opts ...grpc.CallOption) (*ChangeMessageVisibilityResponse, error)
+}
+
+type queueServiceClient struct {
+	cc grpc.ClientConnInterface
+}
+
+func NewQueueServiceClient(cc grpc.ClientConnInterface) QueueServiceClient {
+	return &queueServiceClient{cc}
+}
+
+func (c *queueServiceClient) SendMessage(ctx context.Context, in *SendMessageRequest, opts ...grpc.CallOption) (*SendMessageResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(SendMessageResponse)
+	err := c.cc.Invoke(ctx, QueueService_SendMessage_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *queueServiceClient) SendMessageBatch(ctx context.Context, in *SendMessageBatchRequest, opts ...grpc.CallOption) (*SendMessageBatchResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(SendMessageBatchResponse)
+	err := c.cc.Invoke(ctx, QueueService_SendMessageBatch_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *queueServiceClient) ReceiveMessage(ctx context.Context, in *ReceiveMessageRequest, opts ...grpc.CallOption) (*ReceiveMessageResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(ReceiveMessageResponse)
+	err := c.cc.Invoke(ctx, QueueService_ReceiveMessage_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *queueServiceClient) DeleteMessage(ctx context.Context, in *DeleteMessageRequest, opts ...grpc.CallOption) (*DeleteMessageResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(DeleteMessageResponse)
+	err := c.cc.Invoke(ctx, QueueService_DeleteMessage_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *queueServiceClient) DeleteMessageBatch(ctx context.Context, in *DeleteMessageBatchRequest, opts ...grpc.CallOption) (*DeleteMessageBatchResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(DeleteMessageBatchResponse)
+	err := c.cc.Invoke(ctx, QueueService_DeleteMessageBatch_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *queueServiceClient) ChangeMessageVisibility(ctx context.Context, in *ChangeMessageVisibilityRequest, opts ...grpc.CallOption) (*ChangeMessageVisibilityResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(ChangeMessageVisibilityResponse)
+	err := c.cc.Invoke(ctx, QueueService_ChangeMessageVisibility_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// QueueServiceServer is the server API for QueueService service.
+// All implementations must embed UnimplementedQueueServiceServer
+// for forward compatibility.
+//
+// QueueService is the message-plane API for the Kafka Management Service.
+// Queue lifecycle (create/delete/configure) is managed via the Queue CRD,
+// not this service — see design.md §2a/§2b in kafaka_management_service.
+type QueueServiceServer interface {
+	SendMessage(context.Context, *SendMessageRequest) (*SendMessageResponse, error)
+	SendMessageBatch(context.Context, *SendMessageBatchRequest) (*SendMessageBatchResponse, error)
+	ReceiveMessage(context.Context, *ReceiveMessageRequest) (*ReceiveMessageResponse, error)
+	DeleteMessage(context.Context, *DeleteMessageRequest) (*DeleteMessageResponse, error)
+	DeleteMessageBatch(context.Context, *DeleteMessageBatchRequest) (*DeleteMessageBatchResponse, error)
+	ChangeMessageVisibility(context.Context, *ChangeMessageVisibilityRequest) (*ChangeMessageVisibilityResponse, error)
+	mustEmbedUnimplementedQueueServiceServer()
+}
+
+// UnimplementedQueueServiceServer 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 UnimplementedQueueServiceServer struct{}
+
+func (UnimplementedQueueServiceServer) SendMessage(context.Context, *SendMessageRequest) (*SendMessageResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method SendMessage not implemented")
+}
+func (UnimplementedQueueServiceServer) SendMessageBatch(context.Context, *SendMessageBatchRequest) (*SendMessageBatchResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method SendMessageBatch not implemented")
+}
+func (UnimplementedQueueServiceServer) ReceiveMessage(context.Context, *ReceiveMessageRequest) (*ReceiveMessageResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method ReceiveMessage not implemented")
+}
+func (UnimplementedQueueServiceServer) DeleteMessage(context.Context, *DeleteMessageRequest) (*DeleteMessageResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method DeleteMessage not implemented")
+}
+func (UnimplementedQueueServiceServer) DeleteMessageBatch(context.Context, *DeleteMessageBatchRequest) (*DeleteMessageBatchResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method DeleteMessageBatch not implemented")
+}
+func (UnimplementedQueueServiceServer) ChangeMessageVisibility(context.Context, *ChangeMessageVisibilityRequest) (*ChangeMessageVisibilityResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method ChangeMessageVisibility not implemented")
+}
+func (UnimplementedQueueServiceServer) mustEmbedUnimplementedQueueServiceServer() {}
+func (UnimplementedQueueServiceServer) testEmbeddedByValue()                      {}
+
+// UnsafeQueueServiceServer may be embedded to opt out of forward compatibility for this service.
+// Use of this interface is not recommended, as added methods to QueueServiceServer will
+// result in compilation errors.
+type UnsafeQueueServiceServer interface {
+	mustEmbedUnimplementedQueueServiceServer()
+}
+
+func RegisterQueueServiceServer(s grpc.ServiceRegistrar, srv QueueServiceServer) {
+	// If the following call panics, it indicates UnimplementedQueueServiceServer 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(&QueueService_ServiceDesc, srv)
+}
+
+func _QueueService_SendMessage_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(SendMessageRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(QueueServiceServer).SendMessage(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: QueueService_SendMessage_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(QueueServiceServer).SendMessage(ctx, req.(*SendMessageRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _QueueService_SendMessageBatch_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(SendMessageBatchRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(QueueServiceServer).SendMessageBatch(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: QueueService_SendMessageBatch_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(QueueServiceServer).SendMessageBatch(ctx, req.(*SendMessageBatchRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _QueueService_ReceiveMessage_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ReceiveMessageRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(QueueServiceServer).ReceiveMessage(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: QueueService_ReceiveMessage_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(QueueServiceServer).ReceiveMessage(ctx, req.(*ReceiveMessageRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _QueueService_DeleteMessage_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(DeleteMessageRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(QueueServiceServer).DeleteMessage(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: QueueService_DeleteMessage_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(QueueServiceServer).DeleteMessage(ctx, req.(*DeleteMessageRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _QueueService_DeleteMessageBatch_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(DeleteMessageBatchRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(QueueServiceServer).DeleteMessageBatch(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: QueueService_DeleteMessageBatch_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(QueueServiceServer).DeleteMessageBatch(ctx, req.(*DeleteMessageBatchRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _QueueService_ChangeMessageVisibility_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ChangeMessageVisibilityRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(QueueServiceServer).ChangeMessageVisibility(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: QueueService_ChangeMessageVisibility_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(QueueServiceServer).ChangeMessageVisibility(ctx, req.(*ChangeMessageVisibilityRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+// QueueService_ServiceDesc is the grpc.ServiceDesc for QueueService service.
+// It's only intended for direct use with grpc.RegisterService,
+// and not to be introspected or modified (even as a copy)
+var QueueService_ServiceDesc = grpc.ServiceDesc{
+	ServiceName: "kafkamgmt.v1.QueueService",
+	HandlerType: (*QueueServiceServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "SendMessage",
+			Handler:    _QueueService_SendMessage_Handler,
+		},
+		{
+			MethodName: "SendMessageBatch",
+			Handler:    _QueueService_SendMessageBatch_Handler,
+		},
+		{
+			MethodName: "ReceiveMessage",
+			Handler:    _QueueService_ReceiveMessage_Handler,
+		},
+		{
+			MethodName: "DeleteMessage",
+			Handler:    _QueueService_DeleteMessage_Handler,
+		},
+		{
+			MethodName: "DeleteMessageBatch",
+			Handler:    _QueueService_DeleteMessageBatch_Handler,
+		},
+		{
+			MethodName: "ChangeMessageVisibility",
+			Handler:    _QueueService_ChangeMessageVisibility_Handler,
+		},
+	},
+	Streams:  []grpc.StreamDesc{},
+	Metadata: "kafkamgmt/v1/queue_service.proto",
+}

go.mod

@@ -0,0 +1,17 @@
+module forgejo.riotpiao.homelab.com/rock/kmsvc-proto
+
+go 1.25.0
+
+require (
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0
+	google.golang.org/genproto/googleapis/api v0.0.0-20260618152121-87f3d3e198d3
+	google.golang.org/grpc v1.81.1
+	google.golang.org/protobuf v1.36.11
+)
+
+require (
+	golang.org/x/net v0.51.0 // indirect
+	golang.org/x/sys v0.42.0 // indirect
+	golang.org/x/text v0.36.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20260610212136-7ab31c22f7ad // indirect
+)

go.sum

@@ -0,0 +1,42 @@
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+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/grpc-ecosystem/grpc-gateway/v2 v2.29.0 h1:5VipnvEpbqr2gA2VbM+nYVbkIF28c5ZQfqCBQ5g2xfk=
+github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0/go.mod h1:Hyl3n6Twe1hvtd9XUXDec4pTvgMSEixRuQKPTMH2bNs=
+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=
+golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo=
+golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y=
+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.36.0 h1:JfKh3XmcRPqZPKevfXVpI1wXPTqbkE5f7JA92a55Yxg=
+golang.org/x/text v0.36.0/go.mod h1:NIdBknypM8iqVmPiuco0Dh6P5Jcdk8lJL0CUebqK164=
+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/api v0.0.0-20260618152121-87f3d3e198d3 h1:ctPmKL12ZsoKAlmPUsoW70zEDiYF+/H6aLieXxgAU0k=
+google.golang.org/genproto/googleapis/api v0.0.0-20260618152121-87f3d3e198d3/go.mod h1:Z4WJ5pJOYWFWcHEQUelD5QaZDknIQkpIL/+fyJOT9+A=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260610212136-7ab31c22f7ad h1:45WmJvIV6C2+O/jjLkPUH+F3aOj/1miDoU2DD0+NWbg=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260610212136-7ab31c22f7ad/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=

proto/kafkamgmt/v1/queue_service.proto

@@ -5,10 +5,9 @@ package kafkamgmt.v1;
 import "google/api/annotations.proto";
 import "google/protobuf/timestamp.proto";
 
-// go_package is a neutral placeholder. Each consumer (kafaka_management_service,
+// Generated Go code is committed to this repo under gen/kafkamgmt/v1 and
-// kmsvc-sdk, ...) generates into its own module path and MUST override this via
+// consumed directly via `go get`. Consumers do not run buf generate themselves.
-// buf.gen.yaml managed-mode `override`, not by forking this file.
+option go_package = "forgejo.riotpiao.homelab.com/rock/kmsvc-proto/gen/kafkamgmt/v1;kafkamgmtv1";
-option go_package = "github.com/rockliang/kmsvc-proto/gen/kafkamgmt/v1;kafkamgmtv1";
 
 // QueueService is the message-plane API for the Kafka Management Service.
 // Queue lifecycle (create/delete/configure) is managed via the Queue CRD,