commit c42a5df536ca12f8689f14f56fe3632aa830d6c3 Author: riotpiaole <19826264+Riotpiaole@users.noreply.github.com> Date: Sun Jun 21 19:05:13 2026 -0700 feat: extract kafkamgmt.v1 proto contract from kafaka_management_service Single source of truth for the wire contract shared by the server, kmsvc-sdk, and kmsvc-cli. Ships proto source + buf lint/breaking-change config only; consumers submodule this repo and codegen independently with their own go_package override (see README.md). diff --git a/.forgejo/workflows/ci.yaml b/.forgejo/workflows/ci.yaml new file mode 100644 index 0000000..829ec85 --- /dev/null +++ b/.forgejo/workflows/ci.yaml @@ -0,0 +1,22 @@ +name: ci + +on: + push: + pull_request: + +jobs: + buf: + runs-on: docker + container: + image: bufbuild/buf:latest + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 + + - name: buf lint + run: buf lint + + - name: buf breaking (against main) + run: buf breaking --against '.git#branch=main' + if: github.ref != 'refs/heads/main' diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..fe9aea7 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,7 @@ +# Changelog + +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) + +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. diff --git a/PLAN.md b/PLAN.md new file mode 100644 index 0000000..e8ed13e --- /dev/null +++ b/PLAN.md @@ -0,0 +1,51 @@ +# kmsvc-proto — shared proto contract implementation plan + +## Context + +Third sibling repo alongside `kafaka_management_service` (server), `kmsvc-sdk` (Go client SDK), and `kmsvc-cli` (CLI). Holds the `kafkamgmt.v1` proto definitions as the single source of truth, so the server and the SDK don't generate code from two independently-drifting copies of the same `.proto` file. + +**Scope is intentionally narrow**: this repo ships `.proto` source + buf lint/breaking-change config only — no generated code, no language-specific runtime deps. Each consumer (server, SDK) submodules this repo and runs its own `buf generate` into its own internal package, using its own pinned plugin/runtime versions. This avoids forcing the server and the SDK to agree on the same generated-code/runtime version (protobuf-go, grpc-gateway) — only the wire contract is shared, not the generated artifacts. + +## Repo layout + +``` +kmsvc-proto/ + buf.yaml # lint (DEFAULT) + breaking-change (FILE) rules, same as today's server config + proto/kafkamgmt/v1/queue_service.proto + CHANGELOG.md # human-readable log of wire-contract changes, one entry per tag + .forgejo/workflows/ci.yaml # buf lint + buf breaking against the previous tag +``` + +No `buf.gen.yaml` here — codegen output/plugins are each consumer's concern, not this repo's. + +## Implementation steps + +### Step 1 — Seed the repo +- Move `kafaka_management_service/proto/kafkamgmt/v1/queue_service.proto` and `buf.yaml` into this repo verbatim (git history doesn't follow across repos; note the origin commit in the first commit message for traceability). +- **Verify**: `buf lint` and `buf build` run clean from this repo's root. + +### Step 2 — Breaking-change gate +- Forgejo Actions workflow: on every PR, `buf breaking --against '.git#branch=main'` so an accidental incompatible field/RPC change is caught before merge, not discovered by a downstream consumer. +- **Verify**: a real PR introducing a breaking change (e.g. renumbering a field) fails the check; a compatible addition (new optional field) passes. + +### Step 3 — Tagging convention +- Semantic version tags (`v1.0.0`, `v1.1.0`, ...): **minor** bump for additive/compatible changes, **major** bump for anything `buf breaking` would have flagged (only ever done deliberately, never silently). `CHANGELOG.md` gets one entry per tag. +- **Verify**: tag `v1.0.0` once Step 1+2 are in place and the current contract (the 6 data-plane RPCs already defined) is confirmed stable. + +### Step 4 — Wire into the server repo +- In `kafaka_management_service`: replace the in-repo `proto/` directory with a git submodule at the same path, pointed at this repo's `v1.0.0` tag; `buf.gen.yaml` and the existing `make proto`/`buf generate` flow are otherwise unchanged (still generates into `internal/api/v1`). +- **Verify**: `git submodule update --init && buf generate && go build ./...` in the server repo produces identical generated output to what's committed today (diff should be empty other than the submodule pointer file). + +### Step 5 — Wire into kmsvc-sdk +- `kmsvc-sdk` adds this repo as a submodule at `proto/`, pinned to the same tag the server currently uses, with its own `buf.gen.yaml` generating into `internal/genapi/` (per `kmsvc-sdk/PLAN.md` step 1). +- **Verify**: both repos' generated code, despite being produced independently, satisfy the same wire format — confirmed implicitly by an SDK-vs-server integration test (SDK client talks to a real running server) once both exist. + +## Acceptance criteria +- [ ] `buf lint`/`buf build` pass from a clean checkout. +- [ ] `buf breaking` CI gate blocks incompatible changes on PRs, verified by a real failing PR. +- [ ] `v1.0.0` tag exists and matches the contract currently live in `kafaka_management_service`. +- [ ] Server repo's submodule swap (step 4) produces byte-identical generated code to its current committed output. +- [ ] `kmsvc-sdk`'s codegen against the same tag succeeds (step 5 of this plan / step 1 of `kmsvc-sdk/PLAN.md`). + +## Sequencing note +Do this repo's steps 1–3 before starting `kmsvc-sdk` step 1 (SDK needs something to submodule). Step 4 (migrating the server repo) can happen any time after step 3 — it's a refactor of already-working code, not a blocker for the SDK/CLI work, so it can be deferred or done opportunistically without holding up `kmsvc-sdk`/`kmsvc-cli` progress. diff --git a/README.md b/README.md new file mode 100644 index 0000000..d3b7e4b --- /dev/null +++ b/README.md @@ -0,0 +1,32 @@ +# kmsvc-proto + +Shared `kafkamgmt.v1` proto contract 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. + +## Consuming this repo + +```bash +git submodule add https://forgejo.riotpiao.homelab.com/rock/kmsvc-proto.git proto +git -C proto checkout v1.0.0 # pin to a tag, don't track a moving branch +``` + +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`: + +```yaml +version: v2 +managed: + enabled: true + override: + - file_option: go_package + value: github.com///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). diff --git a/buf.lock b/buf.lock new file mode 100644 index 0000000..8447589 --- /dev/null +++ b/buf.lock @@ -0,0 +1,6 @@ +# Generated by buf. DO NOT EDIT. +version: v2 +deps: + - name: buf.build/googleapis/googleapis + commit: c17df5b2beca46928cc87d5656bd5343 + digest: b5:648a01e0170d4512dea7d564016165decd1ed6e34bef79fe54753e51ad7e27545709ad9157d7551270147d551155c595a2fb0bf5bb33b1c83040ddbce915c604 diff --git a/buf.yaml b/buf.yaml new file mode 100644 index 0000000..4bd5cb5 --- /dev/null +++ b/buf.yaml @@ -0,0 +1,11 @@ +version: v2 +modules: + - path: proto +deps: + - buf.build/googleapis/googleapis +lint: + use: + - STANDARD +breaking: + use: + - FILE diff --git a/proto/kafkamgmt/v1/queue_service.proto b/proto/kafkamgmt/v1/queue_service.proto new file mode 100644 index 0000000..20ad1d5 --- /dev/null +++ b/proto/kafkamgmt/v1/queue_service.proto @@ -0,0 +1,150 @@ +syntax = "proto3"; + +package kafkamgmt.v1; + +import "google/api/annotations.proto"; +import "google/protobuf/timestamp.proto"; + +// go_package is a neutral placeholder. Each consumer (kafaka_management_service, +// kmsvc-sdk, ...) generates into its own module path and MUST override this via +// buf.gen.yaml managed-mode `override`, not by forking this file. +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, +// not this service — see design.md §2a/§2b in kafaka_management_service. +service QueueService { + rpc SendMessage(SendMessageRequest) returns (SendMessageResponse) { + option (google.api.http) = { + post: "/v1/queues/{queue_name}/messages" + body: "*" + }; + } + + rpc SendMessageBatch(SendMessageBatchRequest) returns (SendMessageBatchResponse) { + option (google.api.http) = { + post: "/v1/queues/{queue_name}/messages:batch" + body: "*" + }; + } + + rpc ReceiveMessage(ReceiveMessageRequest) returns (ReceiveMessageResponse) { + option (google.api.http) = { + get: "/v1/queues/{queue_name}/messages" + }; + } + + rpc DeleteMessage(DeleteMessageRequest) returns (DeleteMessageResponse) { + option (google.api.http) = { + delete: "/v1/queues/{queue_name}/messages/{receipt_handle}" + }; + } + + rpc DeleteMessageBatch(DeleteMessageBatchRequest) returns (DeleteMessageBatchResponse) { + option (google.api.http) = { + post: "/v1/queues/{queue_name}/messages:batchDelete" + body: "*" + }; + } + + rpc ChangeMessageVisibility(ChangeMessageVisibilityRequest) returns (ChangeMessageVisibilityResponse) { + option (google.api.http) = { + patch: "/v1/queues/{queue_name}/messages/{receipt_handle}" + body: "*" + }; + } +} + +message MessageAttributes { + map values = 1; +} + +message SendMessageRequest { + string queue_name = 1; + bytes message_body = 2; + MessageAttributes message_attributes = 3; + string message_group_id = 4; // FIFO only + string message_deduplication_id = 5; // FIFO only + int32 delay_seconds = 6; +} + +message SendMessageResponse { + string message_id = 1; + string sequence_number = 2; // FIFO only +} + +message SendMessageBatchEntry { + string id = 1; + bytes message_body = 2; + MessageAttributes message_attributes = 3; + string message_group_id = 4; + string message_deduplication_id = 5; + int32 delay_seconds = 6; +} + +message SendMessageBatchRequest { + string queue_name = 1; + repeated SendMessageBatchEntry entries = 2; +} + +message BatchResultEntry { + string id = 1; + string message_id = 2; + string error = 3; +} + +message SendMessageBatchResponse { + repeated BatchResultEntry successful = 1; + repeated BatchResultEntry failed = 2; +} + +message ReceiveMessageRequest { + string queue_name = 1; + int32 max_number_of_messages = 2; // <= 10 + int32 wait_time_seconds = 3; // 0-20, long-poll + int32 visibility_timeout_seconds = 4; // override +} + +message Message { + string message_id = 1; + string receipt_handle = 2; + bytes body = 3; + MessageAttributes attributes = 4; + int32 receive_count = 5; + string message_group_id = 6; + google.protobuf.Timestamp enqueued_at = 7; +} + +message ReceiveMessageResponse { + repeated Message messages = 1; +} + +message DeleteMessageRequest { + string queue_name = 1; + string receipt_handle = 2; +} + +message DeleteMessageResponse {} + +message DeleteMessageBatchEntry { + string id = 1; + string receipt_handle = 2; +} + +message DeleteMessageBatchRequest { + string queue_name = 1; + repeated DeleteMessageBatchEntry entries = 2; +} + +message DeleteMessageBatchResponse { + repeated BatchResultEntry successful = 1; + repeated BatchResultEntry failed = 2; +} + +message ChangeMessageVisibilityRequest { + string queue_name = 1; + string receipt_handle = 2; + int32 visibility_timeout_seconds = 3; +} + +message ChangeMessageVisibilityResponse {}