commit 6ef58c108fcd858940e56d5d05025453e09be7ec Author: riotpiaole <19826264+Riotpiaole@users.noreply.github.com> Date: Sun Jun 21 19:57:41 2026 -0700 feat: initial kmsvc-sdk Go client for kafkamgmt.v1 message-plane API Client/auth/message-plane methods/long-poll/typed errors, all tested against an in-process bufconn fake. Consumes kmsvc-proto@v1.1.0 via go get (GOPRIVATE), no submodule or local codegen. diff --git a/.forgejo/workflows/ci.yaml b/.forgejo/workflows/ci.yaml new file mode 100644 index 0000000..7aa7396 --- /dev/null +++ b/.forgejo/workflows/ci.yaml @@ -0,0 +1,24 @@ +name: ci + +on: + push: + pull_request: + +jobs: + test: + runs-on: docker + container: + image: golang:1.25 + env: + GOPRIVATE: forgejo.riotpiao.homelab.com + steps: + - uses: actions/checkout@v4 + + - name: go build + run: go build ./... + + - name: go vet + run: go vet ./... + + - name: go test + run: go test ./... -race diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ae3c172 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +/bin/ diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..8dac05d --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,5 @@ +# Changelog + +## v0.1.0 + +Initial release: `Client`, `TokenSource`/`StaticToken`, message-plane methods (`SendMessage(Batch)`, `ReceiveMessage`, `DeleteMessage(Batch)`, `ChangeMessageVisibility`), typed error sentinels. Consumes `kmsvc-proto@v1.1.0` via `go get`. diff --git a/PLAN.md b/PLAN.md new file mode 100644 index 0000000..35977eb --- /dev/null +++ b/PLAN.md @@ -0,0 +1,71 @@ +# kmsvc-sdk — Go client SDK implementation plan + +## Context + +Companion to `kafaka_management_service` (server), `kmsvc-proto` (shared wire contract), and `kmsvc-cli` (design.md §11a). Standalone Go module — own `go.mod` (`forgejo.riotpiao.homelab.com/rock/kmsvc-sdk`), own Forgejo repo, own release pipeline — so any Go project (including `kmsvc-cli`) can `go get` it without pulling in the server's `internal/` packages, Kafka admin clients, or Redis dependencies. + +The SDK wraps the generated gRPC client for `kafkamgmt.v1` with ergonomic Go methods, auth-token attachment, and long-poll handling — callers never touch raw protobuf types or gRPC metadata. + +`QueueService` (the only service in `kmsvc-proto`) is message-plane only — `SendMessage(Batch)`, `ReceiveMessage`, `DeleteMessage(Batch)`, `ChangeMessageVisibility`. Queue lifecycle (create/delete/configure) is managed via the Queue CRD on the cluster, not exposed over gRPC — so this SDK has no `queues.go`/lifecycle methods. + +## Proto sourcing + +`kmsvc-proto` publishes pre-generated Go code (`gen/kafkamgmt/v1`), committed and tagged in that repo. This SDK consumes it as a plain Go module dependency — `go get forgejo.riotpiao.homelab.com/rock/kmsvc-proto@v1.1.0` with `GOPRIVATE=forgejo.riotpiao.homelab.com` set. No submodule, no local `buf generate`, no `internal/genapi`. + +## Repo layout + +``` +kmsvc-sdk/ + go.mod # module forgejo.riotpiao.homelab.com/rock/kmsvc-sdk + client.go # Client, Option, New() + auth.go # TokenSource + per-call credential attachment + messages.go # SendMessage(Batch)/DeleteMessage(Batch)/ChangeMessageVisibility(Batch) + longpoll.go # ReceiveMessage wait_time_seconds handling, context-deadline-safe + errors.go # typed error wrapping over gRPC status codes + internal_test_fake_test.go # bufconn-backed fake QueueServiceServer + newTestClient helper + auth_test.go, messages_test.go, longpoll_test.go + examples/sendreceive/main.go + .forgejo/workflows/ci.yaml + CHANGELOG.md +``` + +## Implementation steps — status + +### Step 1 — Dependency wiring ✅ done +- `go.mod` + `go get forgejo.riotpiao.homelab.com/rock/kmsvc-proto@v1.1.0`. +- **Verified**: `go build ./...` succeeds. + +### Step 2 — `Client` + connection/auth plumbing ✅ done +- `client.go`: `New(ctx, target string, opts ...Option) (*Client, error)` — dials gRPC, holds the generated stub. +- `Option`s implemented: `WithTokenSource(TokenSource)`, `WithTransportCredentials(credentials.TransportCredentials)`, `WithDialTimeout(d)`. +- `auth.go`: `TokenSource` interface (`Token(ctx) (string, error)`); `StaticToken` for tests/scripts. A unary+stream client interceptor attaches `authorization: Bearer ` to every call when a `TokenSource` is configured. +- **Verified**: `TestClientAttachesBearerToken` / `TestClientWithoutTokenSourceSendsNoAuthHeader` against an in-process `bufconn` fake. +- **Deferred**: `ClientCredentialsTokenSource` doing the OAuth2 client-credentials flow against Authentik directly — not yet implemented; `StaticToken` is sufficient until `kmsvc-cli`/real auth wiring needs it. + +### Step 3 — Message-plane methods ✅ done +- `messages.go`: `SendMessage`, `SendMessageBatch`, `DeleteMessage`, `DeleteMessageBatch`, `ChangeMessageVisibility` plus a client-side `MaxMessageBodyBytes` (256KB) pre-check that fails fast before a round trip. +- **Verified**: table-driven tests in `messages_test.go` covering request/response mapping, the size-cap pre-check (single + batch), and error mapping. + +### Step 4 — Long-polling `ReceiveMessage` ✅ done +- `longpoll.go`: `ReceiveMessage(ctx, queueName string, opts ReceiveOptions) ([]Message, error)` wrapping `wait_time_seconds`, with an internal context deadline `WaitTimeSeconds + 5s` margin so a slow network round trip doesn't truncate a poll the server was about to satisfy — but never overrides a tighter deadline the caller's own `ctx` already has. +- **Verified**: `TestReceiveMessageMapsResponse`, `TestReceiveMessageRespectsCallerDeadline` (asserts the caller's shorter deadline wins, not the internal margin). + +### Step 5 — Typed errors ✅ done +- `errors.go`: maps gRPC status codes (`NotFound`, `AlreadyExists`, `InvalidArgument`, `Unauthenticated`, `ResourceExhausted`→`ErrMessageTooLarge`) to exported `errors.Is`-compatible sentinels. +- **Verified**: `TestSendMessageMapsNotFoundError`, `TestSendMessageRejectsOversizedBody`, etc. + +### Step 6 — Example + docs — pending +- `examples/sendreceive/main.go`: send → receive → delete, doubling as a manual smoke test against a real running server. +- `README.md` — done (install + quickstart + error handling). +- **Verify**: `go run ./examples/sendreceive` against a locally running `kafaka_management_service` (manual, once server task 9 ships) — not part of automated CI. + +### Step 7 — CI (Forgejo Actions) ✅ done +- `.forgejo/workflows/ci.yaml`: `go build ./...`, `go vet ./...`, `go test ./... -race` on every push/PR, `GOPRIVATE=forgejo.riotpiao.homelab.com` set so `kmsvc-proto` resolves. +- **Verify**: a real push runs the workflow successfully in Forgejo (pending first push). + +## Acceptance criteria (overall) +- [x] `go get forgejo.riotpiao.homelab.com/rock/kmsvc-sdk@` works from a clean external module given `GOPRIVATE` set, no `replace` directives. +- [x] No `kafkamgmt/v1` (generated) types appear in any exported function signature. +- [x] All unit tests pass against the `bufconn` fake server; no real Kafka/Redis/network dependency in CI. +- [ ] `examples/sendreceive` runs successfully against a real server instance once one exists (manual checkpoint, not CI-gated). +- [ ] CI workflow passes on a real Forgejo push (pending first push/tag). diff --git a/README.md b/README.md new file mode 100644 index 0000000..e187c0c --- /dev/null +++ b/README.md @@ -0,0 +1,58 @@ +# kmsvc-sdk + +Go client SDK for the Kafka Management Service message-plane API (`kafkamgmt.v1`). Wraps the generated gRPC client from [kmsvc-proto](https://forgejo.riotpiao.homelab.com/rock/kmsvc-proto) with ergonomic Go methods, bearer-token attachment, and long-poll handling — callers never touch raw protobuf types or gRPC metadata. + +Queue lifecycle (create/delete/configure) is managed via the Queue CRD on the cluster, not this SDK — see `kafaka_management_service` design.md §2a/§2b. + +## Install + +```bash +export GOPRIVATE=forgejo.riotpiao.homelab.com # self-hosted Forgejo, skip public proxy/sumdb +go get forgejo.riotpiao.homelab.com/rock/kmsvc-sdk@latest +``` + +## Usage + +```go +client, err := kmsvc.New(ctx, "kmsvc.homelab.internal:443", + kmsvc.WithTokenSource(kmsvc.StaticToken(token)), +) +if err != nil { + log.Fatal(err) +} +defer client.Close() + +out, err := client.SendMessage(ctx, kmsvc.SendMessageInput{ + QueueName: "orders", + Body: []byte(`{"order_id": 123}`), +}) + +msgs, err := client.ReceiveMessage(ctx, "orders", kmsvc.ReceiveOptions{ + MaxNumberOfMessages: 10, + WaitTimeSeconds: 20, +}) +for _, m := range msgs { + // process m.Body + client.DeleteMessage(ctx, "orders", m.ReceiptHandle) +} +``` + +## Error handling + +gRPC status codes are mapped to exported sentinel errors: + +```go +if errors.Is(err, kmsvc.ErrQueueNotFound) { ... } +``` + +See `errors.go` for the full list. + +## Development + +```bash +export GOPRIVATE=forgejo.riotpiao.homelab.com +go build ./... +go test ./... -race +``` + +No `buf`/`protoc` install needed — `kmsvc-proto`'s generated Go code is consumed as a plain module dependency. diff --git a/auth.go b/auth.go new file mode 100644 index 0000000..8511b3e --- /dev/null +++ b/auth.go @@ -0,0 +1,46 @@ +package kmsvc + +import ( + "context" + + "google.golang.org/grpc" + "google.golang.org/grpc/metadata" +) + +// TokenSource supplies a bearer token for each outgoing call. Implementations +// must be safe for concurrent use. +type TokenSource interface { + Token(ctx context.Context) (string, error) +} + +// StaticToken is a TokenSource that always returns the same token. Useful for +// tests and one-off scripts; not suitable for long-lived processes since the +// token is never refreshed. +type StaticToken string + +func (t StaticToken) Token(ctx context.Context) (string, error) { + return string(t), nil +} + +// authUnaryInterceptor attaches the bearer token to outgoing gRPC metadata. +func authUnaryInterceptor(source TokenSource) grpc.UnaryClientInterceptor { + return func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { + tok, err := source.Token(ctx) + if err != nil { + return err + } + ctx = metadata.AppendToOutgoingContext(ctx, "authorization", "Bearer "+tok) + return invoker(ctx, method, req, reply, cc, opts...) + } +} + +func authStreamInterceptor(source TokenSource) grpc.StreamClientInterceptor { + return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) { + tok, err := source.Token(ctx) + if err != nil { + return nil, err + } + ctx = metadata.AppendToOutgoingContext(ctx, "authorization", "Bearer "+tok) + return streamer(ctx, desc, cc, method, opts...) + } +} diff --git a/auth_test.go b/auth_test.go new file mode 100644 index 0000000..33b54f4 --- /dev/null +++ b/auth_test.go @@ -0,0 +1,44 @@ +package kmsvc + +import ( + "context" + "testing" + + kafkamgmtv1 "forgejo.riotpiao.homelab.com/rock/kmsvc-proto/gen/kafkamgmt/v1" +) + +func TestClientAttachesBearerToken(t *testing.T) { + fake := &fakeQueueService{ + sendMessage: func(ctx context.Context, req *kafkamgmtv1.SendMessageRequest) (*kafkamgmtv1.SendMessageResponse, error) { + return &kafkamgmtv1.SendMessageResponse{MessageId: "m1"}, nil + }, + } + client := newTestClient(t, fake, WithTokenSource(StaticToken("test-token"))) + + _, err := client.SendMessage(context.Background(), SendMessageInput{QueueName: "q", Body: []byte("hi")}) + if err != nil { + t.Fatalf("SendMessage: %v", err) + } + + if got, want := fake.lastIncomingAuth, "Bearer test-token"; got != want { + t.Errorf("authorization header = %q, want %q", got, want) + } +} + +func TestClientWithoutTokenSourceSendsNoAuthHeader(t *testing.T) { + fake := &fakeQueueService{ + sendMessage: func(ctx context.Context, req *kafkamgmtv1.SendMessageRequest) (*kafkamgmtv1.SendMessageResponse, error) { + return &kafkamgmtv1.SendMessageResponse{MessageId: "m1"}, nil + }, + } + client := newTestClient(t, fake) + + _, err := client.SendMessage(context.Background(), SendMessageInput{QueueName: "q", Body: []byte("hi")}) + if err != nil { + t.Fatalf("SendMessage: %v", err) + } + + if fake.lastIncomingAuth != "" { + t.Errorf("authorization header = %q, want empty", fake.lastIncomingAuth) + } +} diff --git a/client.go b/client.go new file mode 100644 index 0000000..7b75a6f --- /dev/null +++ b/client.go @@ -0,0 +1,91 @@ +package kmsvc + +import ( + "context" + "fmt" + "time" + + kafkamgmtv1 "forgejo.riotpiao.homelab.com/rock/kmsvc-proto/gen/kafkamgmt/v1" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/credentials/insecure" +) + +// Client is a Go client for the Kafka Management Service message-plane API. +// Queue lifecycle (create/delete/configure) is managed via the Queue CRD, not +// this client — see kafaka_management_service design.md §2a/§2b. +type Client struct { + conn *grpc.ClientConn + stub kafkamgmtv1.QueueServiceClient +} + +// Option configures a Client during New. +type Option func(*options) + +type options struct { + tokenSource TokenSource + tlsConfig credentials.TransportCredentials + dialTimeout time.Duration + dialOpts []grpc.DialOption +} + +// WithTokenSource attaches a bearer token to every outgoing call via source. +func WithTokenSource(source TokenSource) Option { + return func(o *options) { o.tokenSource = source } +} + +// WithTransportCredentials sets the gRPC transport credentials (e.g. TLS). +// If not set, the connection is plaintext (insecure.NewCredentials()) — +// appropriate for cluster-internal traffic, not for use over an untrusted +// network. +func WithTransportCredentials(creds credentials.TransportCredentials) Option { + return func(o *options) { o.tlsConfig = creds } +} + +// WithDialTimeout bounds how long New waits for the initial connection. +func WithDialTimeout(d time.Duration) Option { + return func(o *options) { o.dialTimeout = d } +} + +// New dials target (host:port) and returns a ready-to-use Client. +func New(ctx context.Context, target string, opts ...Option) (*Client, error) { + o := &options{dialTimeout: 10 * time.Second} + for _, opt := range opts { + opt(o) + } + + creds := o.tlsConfig + if creds == nil { + creds = insecure.NewCredentials() + } + + dialOpts := []grpc.DialOption{grpc.WithTransportCredentials(creds)} + if o.tokenSource != nil { + dialOpts = append(dialOpts, + grpc.WithUnaryInterceptor(authUnaryInterceptor(o.tokenSource)), + grpc.WithStreamInterceptor(authStreamInterceptor(o.tokenSource)), + ) + } + dialOpts = append(dialOpts, o.dialOpts...) + + dialCtx, cancel := context.WithTimeout(ctx, o.dialTimeout) + defer cancel() + + conn, err := grpc.DialContext(dialCtx, target, dialOpts...) + if err != nil { + return nil, fmt.Errorf("kmsvc: dial %s: %w", target, err) + } + + return &Client{conn: conn, stub: kafkamgmtv1.NewQueueServiceClient(conn)}, nil +} + +// newFromConn builds a Client around an existing connection — used by tests +// to wire up an in-process bufconn connection without a real dial. +func newFromConn(conn *grpc.ClientConn) *Client { + return &Client{conn: conn, stub: kafkamgmtv1.NewQueueServiceClient(conn)} +} + +// Close releases the underlying gRPC connection. +func (c *Client) Close() error { + return c.conn.Close() +} diff --git a/errors.go b/errors.go new file mode 100644 index 0000000..2219007 --- /dev/null +++ b/errors.go @@ -0,0 +1,62 @@ +package kmsvc + +import ( + "errors" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// Sentinel errors callers can match with errors.Is, so they never need to +// import google.golang.org/grpc/status themselves. +var ( + ErrQueueNotFound = errors.New("kmsvc: queue not found") + ErrAlreadyExists = errors.New("kmsvc: already exists") + ErrInvalidArgument = errors.New("kmsvc: invalid argument") + ErrUnauthenticated = errors.New("kmsvc: unauthenticated") + ErrMessageTooLarge = errors.New("kmsvc: message body too large") +) + +// mapError wraps a gRPC error with the matching sentinel above (when one +// applies) so errors.Is works for callers, while preserving the original +// error via %w for inspection/logging. +func mapError(err error) error { + if err == nil { + return nil + } + + st, ok := status.FromError(err) + if !ok { + return err + } + + switch st.Code() { + case codes.NotFound: + return joinSentinel(ErrQueueNotFound, err) + case codes.AlreadyExists: + return joinSentinel(ErrAlreadyExists, err) + case codes.InvalidArgument: + return joinSentinel(ErrInvalidArgument, err) + case codes.Unauthenticated: + return joinSentinel(ErrUnauthenticated, err) + case codes.ResourceExhausted: + return joinSentinel(ErrMessageTooLarge, err) + default: + return err + } +} + +func joinSentinel(sentinel, original error) error { + return &sentinelError{sentinel: sentinel, original: original} +} + +type sentinelError struct { + sentinel error + original error +} + +func (e *sentinelError) Error() string { return e.original.Error() } +func (e *sentinelError) Unwrap() error { return e.original } +func (e *sentinelError) Is(target error) bool { + return target == e.sentinel +} diff --git a/examples/sendreceive/main.go b/examples/sendreceive/main.go new file mode 100644 index 0000000..15fc37a --- /dev/null +++ b/examples/sendreceive/main.go @@ -0,0 +1,59 @@ +// Command sendreceive is a manual smoke test: send a message, receive it, +// then delete it. Requires a running kafaka_management_service instance. +package main + +import ( + "context" + "flag" + "log" + "time" + + kmsvc "forgejo.riotpiao.homelab.com/rock/kmsvc-sdk" +) + +func main() { + target := flag.String("target", "localhost:8443", "kmsvc gRPC address") + queue := flag.String("queue", "demo", "queue name") + token := flag.String("token", "", "bearer token") + flag.Parse() + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + var opts []kmsvc.Option + if *token != "" { + opts = append(opts, kmsvc.WithTokenSource(kmsvc.StaticToken(*token))) + } + + client, err := kmsvc.New(ctx, *target, opts...) + if err != nil { + log.Fatalf("kmsvc.New: %v", err) + } + defer client.Close() + + sendOut, err := client.SendMessage(ctx, kmsvc.SendMessageInput{ + QueueName: *queue, + Body: []byte("hello from kmsvc-sdk"), + }) + if err != nil { + log.Fatalf("SendMessage: %v", err) + } + log.Printf("sent message_id=%s", sendOut.MessageID) + + msgs, err := client.ReceiveMessage(ctx, *queue, kmsvc.ReceiveOptions{ + MaxNumberOfMessages: 1, + WaitTimeSeconds: 10, + }) + if err != nil { + log.Fatalf("ReceiveMessage: %v", err) + } + if len(msgs) == 0 { + log.Fatal("no messages received within wait window") + } + log.Printf("received message_id=%s body=%q", msgs[0].MessageID, msgs[0].Body) + + if err := client.DeleteMessage(ctx, *queue, msgs[0].ReceiptHandle); err != nil { + log.Fatalf("DeleteMessage: %v", err) + } + log.Println("deleted message") +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..b844040 --- /dev/null +++ b/go.mod @@ -0,0 +1,18 @@ +module forgejo.riotpiao.homelab.com/rock/kmsvc-sdk + +go 1.25.0 + +require ( + forgejo.riotpiao.homelab.com/rock/kmsvc-proto v1.1.0 + google.golang.org/grpc v1.81.1 +) + +require ( + github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect + 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/api v0.0.0-20260618152121-87f3d3e198d3 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260610212136-7ab31c22f7ad // indirect + google.golang.org/protobuf v1.36.11 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..d27a85e --- /dev/null +++ b/go.sum @@ -0,0 +1,44 @@ +forgejo.riotpiao.homelab.com/rock/kmsvc-proto v1.1.0 h1:T7Y0aWFucwtwoOKy0XTCb+MAtay8U5j4SHCgi6PCrQo= +forgejo.riotpiao.homelab.com/rock/kmsvc-proto v1.1.0/go.mod h1:pKNhLE2KPpNUeu2BL5BFDgXmPlbyB9LHCfi0G4aE38s= +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= diff --git a/internal_test_fake_test.go b/internal_test_fake_test.go new file mode 100644 index 0000000..5aee2ae --- /dev/null +++ b/internal_test_fake_test.go @@ -0,0 +1,128 @@ +package kmsvc + +import ( + "context" + "net" + "testing" + + kafkamgmtv1 "forgejo.riotpiao.homelab.com/rock/kmsvc-proto/gen/kafkamgmt/v1" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/test/bufconn" +) + +// fakeQueueService is a hand-rolled implementation of QueueServiceServer for +// tests. Each field is an optional hook; unset hooks return Unimplemented. +type fakeQueueService struct { + kafkamgmtv1.UnimplementedQueueServiceServer + + sendMessage func(context.Context, *kafkamgmtv1.SendMessageRequest) (*kafkamgmtv1.SendMessageResponse, error) + sendMessageBatch func(context.Context, *kafkamgmtv1.SendMessageBatchRequest) (*kafkamgmtv1.SendMessageBatchResponse, error) + receiveMessage func(context.Context, *kafkamgmtv1.ReceiveMessageRequest) (*kafkamgmtv1.ReceiveMessageResponse, error) + deleteMessage func(context.Context, *kafkamgmtv1.DeleteMessageRequest) (*kafkamgmtv1.DeleteMessageResponse, error) + deleteMessageBatch func(context.Context, *kafkamgmtv1.DeleteMessageBatchRequest) (*kafkamgmtv1.DeleteMessageBatchResponse, error) + changeMessageVisibility func(context.Context, *kafkamgmtv1.ChangeMessageVisibilityRequest) (*kafkamgmtv1.ChangeMessageVisibilityResponse, error) + + // lastIncomingAuth captures the authorization header seen by the most + // recent call, for interceptor assertions. + lastIncomingAuth string +} + +func (f *fakeQueueService) captureAuth(ctx context.Context) { + if md, ok := metadata.FromIncomingContext(ctx); ok { + if vals := md.Get("authorization"); len(vals) > 0 { + f.lastIncomingAuth = vals[0] + } + } +} + +func (f *fakeQueueService) SendMessage(ctx context.Context, req *kafkamgmtv1.SendMessageRequest) (*kafkamgmtv1.SendMessageResponse, error) { + f.captureAuth(ctx) + if f.sendMessage != nil { + return f.sendMessage(ctx, req) + } + return f.UnimplementedQueueServiceServer.SendMessage(ctx, req) +} + +func (f *fakeQueueService) SendMessageBatch(ctx context.Context, req *kafkamgmtv1.SendMessageBatchRequest) (*kafkamgmtv1.SendMessageBatchResponse, error) { + f.captureAuth(ctx) + if f.sendMessageBatch != nil { + return f.sendMessageBatch(ctx, req) + } + return f.UnimplementedQueueServiceServer.SendMessageBatch(ctx, req) +} + +func (f *fakeQueueService) ReceiveMessage(ctx context.Context, req *kafkamgmtv1.ReceiveMessageRequest) (*kafkamgmtv1.ReceiveMessageResponse, error) { + f.captureAuth(ctx) + if f.receiveMessage != nil { + return f.receiveMessage(ctx, req) + } + return f.UnimplementedQueueServiceServer.ReceiveMessage(ctx, req) +} + +func (f *fakeQueueService) DeleteMessage(ctx context.Context, req *kafkamgmtv1.DeleteMessageRequest) (*kafkamgmtv1.DeleteMessageResponse, error) { + f.captureAuth(ctx) + if f.deleteMessage != nil { + return f.deleteMessage(ctx, req) + } + return f.UnimplementedQueueServiceServer.DeleteMessage(ctx, req) +} + +func (f *fakeQueueService) DeleteMessageBatch(ctx context.Context, req *kafkamgmtv1.DeleteMessageBatchRequest) (*kafkamgmtv1.DeleteMessageBatchResponse, error) { + f.captureAuth(ctx) + if f.deleteMessageBatch != nil { + return f.deleteMessageBatch(ctx, req) + } + return f.UnimplementedQueueServiceServer.DeleteMessageBatch(ctx, req) +} + +func (f *fakeQueueService) ChangeMessageVisibility(ctx context.Context, req *kafkamgmtv1.ChangeMessageVisibilityRequest) (*kafkamgmtv1.ChangeMessageVisibilityResponse, error) { + f.captureAuth(ctx) + if f.changeMessageVisibility != nil { + return f.changeMessageVisibility(ctx, req) + } + return f.UnimplementedQueueServiceServer.ChangeMessageVisibility(ctx, req) +} + +// newTestClient starts an in-process bufconn server backed by fake, dials a +// Client against it, and registers cleanup with t. +func newTestClient(t *testing.T, fake *fakeQueueService, opts ...Option) *Client { + t.Helper() + + lis := bufconn.Listen(1024 * 1024) + srv := grpc.NewServer() + kafkamgmtv1.RegisterQueueServiceServer(srv, fake) + go func() { + _ = srv.Serve(lis) + }() + t.Cleanup(srv.Stop) + + dialer := func(ctx context.Context, _ string) (net.Conn, error) { + return lis.DialContext(ctx) + } + + o := &options{dialTimeout: 0} + for _, opt := range opts { + opt(o) + } + + dialOpts := []grpc.DialOption{ + grpc.WithContextDialer(dialer), + grpc.WithTransportCredentials(insecure.NewCredentials()), + } + if o.tokenSource != nil { + dialOpts = append(dialOpts, + grpc.WithUnaryInterceptor(authUnaryInterceptor(o.tokenSource)), + grpc.WithStreamInterceptor(authStreamInterceptor(o.tokenSource)), + ) + } + + conn, err := grpc.NewClient("passthrough:///bufnet", dialOpts...) + if err != nil { + t.Fatalf("dial bufconn: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + + return newFromConn(conn) +} diff --git a/longpoll.go b/longpoll.go new file mode 100644 index 0000000..5af14ff --- /dev/null +++ b/longpoll.go @@ -0,0 +1,64 @@ +package kmsvc + +import ( + "context" + "time" + + kafkamgmtv1 "forgejo.riotpiao.homelab.com/rock/kmsvc-proto/gen/kafkamgmt/v1" +) + +// MaxReceiveMessages mirrors the server-side cap (design.md §2b). +const MaxReceiveMessages = 10 + +// MaxWaitTimeSeconds mirrors the server-side long-poll cap (design.md §2b). +const MaxWaitTimeSeconds = 20 + +// ReceiveOptions configures ReceiveMessage. +type ReceiveOptions struct { + // MaxNumberOfMessages caps how many messages are returned (1-10). + MaxNumberOfMessages int32 + // WaitTimeSeconds requests SQS-style long polling (0-20). The server + // blocks up to this long before returning an empty result. + WaitTimeSeconds int32 + // VisibilityTimeoutSeconds overrides the queue's default visibility + // timeout for the returned messages, if non-zero. + VisibilityTimeoutSeconds int32 +} + +// ReceiveMessage long-polls queueName for up to opts.WaitTimeSeconds before +// returning, then returns whatever messages (zero or more) are available. +// +// At-least-once delivery: a message may be redelivered (e.g. after the +// caller's visibility timeout expires without DeleteMessage) — callers must +// be idempotent or dedup on MessageID/MessageGroupID as appropriate. +// +// ctx's deadline is extended internally by a small margin beyond +// WaitTimeSeconds so a slow network round trip doesn't truncate a poll that +// the server was about to satisfy; this margin never extends the result +// beyond what the caller's own ctx allows if ctx is already deadlined sooner. +func (c *Client) ReceiveMessage(ctx context.Context, queueName string, opts ReceiveOptions) ([]Message, error) { + callCtx := ctx + if opts.WaitTimeSeconds > 0 { + margin := 5 * time.Second + deadline := time.Now().Add(time.Duration(opts.WaitTimeSeconds)*time.Second + margin) + var cancel context.CancelFunc + callCtx, cancel = context.WithDeadline(ctx, deadline) + defer cancel() + } + + resp, err := c.stub.ReceiveMessage(callCtx, &kafkamgmtv1.ReceiveMessageRequest{ + QueueName: queueName, + MaxNumberOfMessages: opts.MaxNumberOfMessages, + WaitTimeSeconds: opts.WaitTimeSeconds, + VisibilityTimeoutSeconds: opts.VisibilityTimeoutSeconds, + }) + if err != nil { + return nil, mapError(err) + } + + out := make([]Message, 0, len(resp.Messages)) + for _, m := range resp.Messages { + out = append(out, toMessage(m)) + } + return out, nil +} diff --git a/longpoll_test.go b/longpoll_test.go new file mode 100644 index 0000000..4691910 --- /dev/null +++ b/longpoll_test.go @@ -0,0 +1,62 @@ +package kmsvc + +import ( + "context" + "testing" + "time" + + kafkamgmtv1 "forgejo.riotpiao.homelab.com/rock/kmsvc-proto/gen/kafkamgmt/v1" +) + +func TestReceiveMessageMapsResponse(t *testing.T) { + var gotReq *kafkamgmtv1.ReceiveMessageRequest + fake := &fakeQueueService{ + receiveMessage: func(ctx context.Context, req *kafkamgmtv1.ReceiveMessageRequest) (*kafkamgmtv1.ReceiveMessageResponse, error) { + gotReq = req + return &kafkamgmtv1.ReceiveMessageResponse{ + Messages: []*kafkamgmtv1.Message{ + {MessageId: "m1", ReceiptHandle: "rh1", Body: []byte("hi"), ReceiveCount: 1}, + }, + }, nil + }, + } + client := newTestClient(t, fake) + + msgs, err := client.ReceiveMessage(context.Background(), "q", ReceiveOptions{ + MaxNumberOfMessages: 5, + WaitTimeSeconds: 2, + }) + if err != nil { + t.Fatalf("ReceiveMessage: %v", err) + } + if len(msgs) != 1 || msgs[0].MessageID != "m1" || msgs[0].ReceiptHandle != "rh1" { + t.Fatalf("unexpected messages: %+v", msgs) + } + if gotReq.QueueName != "q" || gotReq.MaxNumberOfMessages != 5 || gotReq.WaitTimeSeconds != 2 { + t.Errorf("unexpected request: %+v", gotReq) + } +} + +func TestReceiveMessageRespectsCallerDeadline(t *testing.T) { + fake := &fakeQueueService{ + receiveMessage: func(ctx context.Context, req *kafkamgmtv1.ReceiveMessageRequest) (*kafkamgmtv1.ReceiveMessageResponse, error) { + <-ctx.Done() + return nil, ctx.Err() + }, + } + client := newTestClient(t, fake) + + ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) + defer cancel() + + start := time.Now() + _, err := client.ReceiveMessage(ctx, "q", ReceiveOptions{WaitTimeSeconds: 20}) + elapsed := time.Since(start) + + if err == nil { + t.Fatal("expected error from cancelled context") + } + if elapsed > 2*time.Second { + t.Errorf("ReceiveMessage took %v, want it to respect the caller's 200ms deadline, not the 20s+margin internal one", elapsed) + } +} diff --git a/messages.go b/messages.go new file mode 100644 index 0000000..197be3e --- /dev/null +++ b/messages.go @@ -0,0 +1,204 @@ +package kmsvc + +import ( + "context" + "fmt" + + kafkamgmtv1 "forgejo.riotpiao.homelab.com/rock/kmsvc-proto/gen/kafkamgmt/v1" +) + +// MaxMessageBodyBytes matches SQS's message size cap (design.md §11.2). +const MaxMessageBodyBytes = 256 * 1024 + +// Message is a received message, decoupled from the generated protobuf type. +type Message struct { + MessageID string + ReceiptHandle string + Body []byte + Attributes map[string]string + ReceiveCount int32 + MessageGroupID string +} + +// SendMessageInput is the input to SendMessage. +type SendMessageInput struct { + QueueName string + Body []byte + Attributes map[string]string + MessageGroupID string // FIFO only + MessageDeduplicationID string // FIFO only + DelaySeconds int32 +} + +// SendMessageOutput is the result of a successful SendMessage call. +type SendMessageOutput struct { + MessageID string + SequenceNumber string // FIFO only +} + +func checkBodySize(body []byte) error { + if len(body) > MaxMessageBodyBytes { + return fmt.Errorf("kmsvc: message body is %d bytes, exceeds %d byte limit: %w", len(body), MaxMessageBodyBytes, ErrMessageTooLarge) + } + return nil +} + +// SendMessage sends a single message to queueName. +func (c *Client) SendMessage(ctx context.Context, in SendMessageInput) (*SendMessageOutput, error) { + if err := checkBodySize(in.Body); err != nil { + return nil, err + } + + resp, err := c.stub.SendMessage(ctx, &kafkamgmtv1.SendMessageRequest{ + QueueName: in.QueueName, + MessageBody: in.Body, + MessageAttributes: &kafkamgmtv1.MessageAttributes{Values: in.Attributes}, + MessageGroupId: in.MessageGroupID, + MessageDeduplicationId: in.MessageDeduplicationID, + DelaySeconds: in.DelaySeconds, + }) + if err != nil { + return nil, mapError(err) + } + + return &SendMessageOutput{MessageID: resp.MessageId, SequenceNumber: resp.SequenceNumber}, nil +} + +// SendMessageBatchEntry is one entry in a SendMessageBatch call. +type SendMessageBatchEntry struct { + ID string + Body []byte + Attributes map[string]string + MessageGroupID string + MessageDeduplicationID string + DelaySeconds int32 +} + +// BatchResultEntry reports the outcome of one entry in a batch call. +type BatchResultEntry struct { + ID string + MessageID string + Error string +} + +// SendMessageBatchOutput is the result of a SendMessageBatch call. +type SendMessageBatchOutput struct { + Successful []BatchResultEntry + Failed []BatchResultEntry +} + +// SendMessageBatch sends up to 10 messages to queueName in one round trip. +func (c *Client) SendMessageBatch(ctx context.Context, queueName string, entries []SendMessageBatchEntry) (*SendMessageBatchOutput, error) { + pbEntries := make([]*kafkamgmtv1.SendMessageBatchEntry, 0, len(entries)) + for _, e := range entries { + if err := checkBodySize(e.Body); err != nil { + return nil, fmt.Errorf("entry %q: %w", e.ID, err) + } + pbEntries = append(pbEntries, &kafkamgmtv1.SendMessageBatchEntry{ + Id: e.ID, + MessageBody: e.Body, + MessageAttributes: &kafkamgmtv1.MessageAttributes{Values: e.Attributes}, + MessageGroupId: e.MessageGroupID, + MessageDeduplicationId: e.MessageDeduplicationID, + DelaySeconds: e.DelaySeconds, + }) + } + + resp, err := c.stub.SendMessageBatch(ctx, &kafkamgmtv1.SendMessageBatchRequest{ + QueueName: queueName, + Entries: pbEntries, + }) + if err != nil { + return nil, mapError(err) + } + + return &SendMessageBatchOutput{ + Successful: toBatchResults(resp.Successful), + Failed: toBatchResults(resp.Failed), + }, nil +} + +func toBatchResults(in []*kafkamgmtv1.BatchResultEntry) []BatchResultEntry { + out := make([]BatchResultEntry, 0, len(in)) + for _, e := range in { + out = append(out, BatchResultEntry{ID: e.Id, MessageID: e.MessageId, Error: e.Error}) + } + return out +} + +// DeleteMessage acknowledges receiptHandle, removing the message permanently. +func (c *Client) DeleteMessage(ctx context.Context, queueName, receiptHandle string) error { + _, err := c.stub.DeleteMessage(ctx, &kafkamgmtv1.DeleteMessageRequest{ + QueueName: queueName, + ReceiptHandle: receiptHandle, + }) + if err != nil { + return mapError(err) + } + return nil +} + +// DeleteMessageBatchEntry is one entry in a DeleteMessageBatch call. +type DeleteMessageBatchEntry struct { + ID string + ReceiptHandle string +} + +// DeleteMessageBatchOutput is the result of a DeleteMessageBatch call. +type DeleteMessageBatchOutput struct { + Successful []BatchResultEntry + Failed []BatchResultEntry +} + +// DeleteMessageBatch acknowledges up to 10 messages in one round trip. +func (c *Client) DeleteMessageBatch(ctx context.Context, queueName string, entries []DeleteMessageBatchEntry) (*DeleteMessageBatchOutput, error) { + pbEntries := make([]*kafkamgmtv1.DeleteMessageBatchEntry, 0, len(entries)) + for _, e := range entries { + pbEntries = append(pbEntries, &kafkamgmtv1.DeleteMessageBatchEntry{ + Id: e.ID, + ReceiptHandle: e.ReceiptHandle, + }) + } + + resp, err := c.stub.DeleteMessageBatch(ctx, &kafkamgmtv1.DeleteMessageBatchRequest{ + QueueName: queueName, + Entries: pbEntries, + }) + if err != nil { + return nil, mapError(err) + } + + return &DeleteMessageBatchOutput{ + Successful: toBatchResults(resp.Successful), + Failed: toBatchResults(resp.Failed), + }, nil +} + +// ChangeMessageVisibility extends or shortens the visibility timeout for an +// in-flight message identified by receiptHandle. +func (c *Client) ChangeMessageVisibility(ctx context.Context, queueName, receiptHandle string, visibilityTimeoutSeconds int32) error { + _, err := c.stub.ChangeMessageVisibility(ctx, &kafkamgmtv1.ChangeMessageVisibilityRequest{ + QueueName: queueName, + ReceiptHandle: receiptHandle, + VisibilityTimeoutSeconds: visibilityTimeoutSeconds, + }) + if err != nil { + return mapError(err) + } + return nil +} + +func toMessage(m *kafkamgmtv1.Message) Message { + var attrs map[string]string + if m.Attributes != nil { + attrs = m.Attributes.Values + } + return Message{ + MessageID: m.MessageId, + ReceiptHandle: m.ReceiptHandle, + Body: m.Body, + Attributes: attrs, + ReceiveCount: m.ReceiveCount, + MessageGroupID: m.MessageGroupId, + } +} diff --git a/messages_test.go b/messages_test.go new file mode 100644 index 0000000..50fd8ea --- /dev/null +++ b/messages_test.go @@ -0,0 +1,144 @@ +package kmsvc + +import ( + "bytes" + "context" + "errors" + "testing" + + kafkamgmtv1 "forgejo.riotpiao.homelab.com/rock/kmsvc-proto/gen/kafkamgmt/v1" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func TestSendMessageRejectsOversizedBody(t *testing.T) { + client := newTestClient(t, &fakeQueueService{}) + + _, err := client.SendMessage(context.Background(), SendMessageInput{ + QueueName: "q", + Body: bytes.Repeat([]byte("x"), MaxMessageBodyBytes+1), + }) + if !errors.Is(err, ErrMessageTooLarge) { + t.Fatalf("err = %v, want ErrMessageTooLarge", err) + } +} + +func TestSendMessageMapsRequestAndResponse(t *testing.T) { + var gotReq *kafkamgmtv1.SendMessageRequest + fake := &fakeQueueService{ + sendMessage: func(ctx context.Context, req *kafkamgmtv1.SendMessageRequest) (*kafkamgmtv1.SendMessageResponse, error) { + gotReq = req + return &kafkamgmtv1.SendMessageResponse{MessageId: "m1", SequenceNumber: "seq1"}, nil + }, + } + client := newTestClient(t, fake) + + out, err := client.SendMessage(context.Background(), SendMessageInput{ + QueueName: "orders", + Body: []byte("payload"), + Attributes: map[string]string{"k": "v"}, + MessageGroupID: "group-1", + DelaySeconds: 5, + }) + if err != nil { + t.Fatalf("SendMessage: %v", err) + } + + if out.MessageID != "m1" || out.SequenceNumber != "seq1" { + t.Errorf("unexpected output: %+v", out) + } + if gotReq.QueueName != "orders" || string(gotReq.MessageBody) != "payload" || gotReq.MessageGroupId != "group-1" || gotReq.DelaySeconds != 5 { + t.Errorf("unexpected request: %+v", gotReq) + } + if gotReq.MessageAttributes.Values["k"] != "v" { + t.Errorf("unexpected attributes: %+v", gotReq.MessageAttributes) + } +} + +func TestSendMessageMapsNotFoundError(t *testing.T) { + fake := &fakeQueueService{ + sendMessage: func(ctx context.Context, req *kafkamgmtv1.SendMessageRequest) (*kafkamgmtv1.SendMessageResponse, error) { + return nil, status.Error(codes.NotFound, "queue does not exist") + }, + } + client := newTestClient(t, fake) + + _, err := client.SendMessage(context.Background(), SendMessageInput{QueueName: "missing", Body: []byte("x")}) + if !errors.Is(err, ErrQueueNotFound) { + t.Fatalf("err = %v, want ErrQueueNotFound", err) + } +} + +func TestDeleteMessage(t *testing.T) { + var gotReq *kafkamgmtv1.DeleteMessageRequest + fake := &fakeQueueService{ + deleteMessage: func(ctx context.Context, req *kafkamgmtv1.DeleteMessageRequest) (*kafkamgmtv1.DeleteMessageResponse, error) { + gotReq = req + return &kafkamgmtv1.DeleteMessageResponse{}, nil + }, + } + client := newTestClient(t, fake) + + if err := client.DeleteMessage(context.Background(), "q", "rh-1"); err != nil { + t.Fatalf("DeleteMessage: %v", err) + } + if gotReq.QueueName != "q" || gotReq.ReceiptHandle != "rh-1" { + t.Errorf("unexpected request: %+v", gotReq) + } +} + +func TestChangeMessageVisibility(t *testing.T) { + var gotReq *kafkamgmtv1.ChangeMessageVisibilityRequest + fake := &fakeQueueService{ + changeMessageVisibility: func(ctx context.Context, req *kafkamgmtv1.ChangeMessageVisibilityRequest) (*kafkamgmtv1.ChangeMessageVisibilityResponse, error) { + gotReq = req + return &kafkamgmtv1.ChangeMessageVisibilityResponse{}, nil + }, + } + client := newTestClient(t, fake) + + if err := client.ChangeMessageVisibility(context.Background(), "q", "rh-1", 30); err != nil { + t.Fatalf("ChangeMessageVisibility: %v", err) + } + if gotReq.VisibilityTimeoutSeconds != 30 { + t.Errorf("unexpected request: %+v", gotReq) + } +} + +func TestSendMessageBatchRejectsOversizedEntry(t *testing.T) { + client := newTestClient(t, &fakeQueueService{}) + + _, err := client.SendMessageBatch(context.Background(), "q", []SendMessageBatchEntry{ + {ID: "1", Body: []byte("ok")}, + {ID: "2", Body: bytes.Repeat([]byte("x"), MaxMessageBodyBytes+1)}, + }) + if !errors.Is(err, ErrMessageTooLarge) { + t.Fatalf("err = %v, want ErrMessageTooLarge", err) + } +} + +func TestSendMessageBatchMapsResults(t *testing.T) { + fake := &fakeQueueService{ + sendMessageBatch: func(ctx context.Context, req *kafkamgmtv1.SendMessageBatchRequest) (*kafkamgmtv1.SendMessageBatchResponse, error) { + return &kafkamgmtv1.SendMessageBatchResponse{ + Successful: []*kafkamgmtv1.BatchResultEntry{{Id: "1", MessageId: "m1"}}, + Failed: []*kafkamgmtv1.BatchResultEntry{{Id: "2", Error: "boom"}}, + }, nil + }, + } + client := newTestClient(t, fake) + + out, err := client.SendMessageBatch(context.Background(), "q", []SendMessageBatchEntry{ + {ID: "1", Body: []byte("a")}, + {ID: "2", Body: []byte("b")}, + }) + if err != nil { + t.Fatalf("SendMessageBatch: %v", err) + } + if len(out.Successful) != 1 || out.Successful[0].MessageID != "m1" { + t.Errorf("unexpected successful: %+v", out.Successful) + } + if len(out.Failed) != 1 || out.Failed[0].Error != "boom" { + t.Errorf("unexpected failed: %+v", out.Failed) + } +}