gRPC

The grpc package wraps google.golang.org/grpc with Velocity conventions - context helpers for auth/request metadata, structured error constructors, and a sibling HTTP gateway that exposes gRPC services over REST via grpc-gateway.

Import path: github.com/velocitykode/velocity/grpc

Sub-packages:

• grpc/interceptors - bundled Auth, Logging, Recovery interceptor pairs
• grpc/grpcevents - request / stream / server / gateway / panic / auth event types
• grpc/converters - proto Timestamp / Duration and pagination helpers

Setup

A typical layout: .proto files under api/proto/<pkg>/<version>/, generated stubs under api/gen/go/..., service implementations under internal/grpc/services/, server lifecycle in a module under internal/modules/.

vel gen grpc service Foo                   # proto + impl + module (idempotent)
vel gen grpc rpc Foo Hello                 # unary
vel gen grpc rpc Foo Tail   --stream       # server-stream
vel gen grpc rpc Foo Upload --client-stream
vel gen grpc rpc Foo Chat   --bidi
vel gen grpc gen                           # cd api/proto && buf generate

Proto + buf

api/proto/foo/v1/foo.proto:

syntax = "proto3";

package foo.v1;

option go_package = "yourapp/api/gen/go/foo/v1;foov1";

service FooService {
  rpc Hello(HelloRequest) returns (HelloResponse);
}

message HelloRequest  { string name = 1; }
message HelloResponse { string greeting = 1; }

api/proto/buf.yaml:

version: v2
modules:
  - path: .
lint:
  use:
    - STANDARD
breaking:
  use:
    - FILE

api/proto/buf.gen.yaml:

version: v2
plugins:
  - remote: buf.build/protocolbuffers/go
    out: ../gen/go
    opt: paths=source_relative
  - remote: buf.build/grpc/go
    out: ../gen/go
    opt:
      - paths=source_relative
      - require_unimplemented_servers=false

Both files are written by the first vel gen grpc service call and left untouched afterwards. Generate stubs:

vel gen grpc gen      # runs `buf generate` inside api/proto

Service implementation

internal/grpc/services/foo.go:

package services

import (
    "context"

    foopb "yourapp/api/gen/go/foo/v1"
)

type FooService struct {
    foopb.UnimplementedFooServiceServer
}

func NewFooService() *FooService { return &FooService{} }

func (s *FooService) Hello(ctx context.Context, req *foopb.HelloRequest) (*foopb.HelloResponse, error) {
    name := req.GetName()
    if name == "" {
        name = "world"
    }
    return &foopb.HelloResponse{Greeting: "hi " + name}, nil
}

Module

internal/modules/grpc_module.go, as the generator writes it:

package modules

import (
    "context"
    "os"

    // vel:grpc:imports
    foopb "yourapp/api/gen/go/foo/v1"
    "yourapp/internal/grpc/services"

    "github.com/velocitykode/velocity"
    velgrpc "github.com/velocitykode/velocity/grpc"
    googleGrpc "google.golang.org/grpc"
)

type GRPCModule struct {
    server *velgrpc.Server
}

func (p *GRPCModule) Init(s *velocity.Services) error {
    port := os.Getenv("GRPC_PORT")
    if port == "" {
        port = "50051"
    }

    p.server = velgrpc.NewServer(
        velgrpc.WithPort(port),
        velgrpc.WithLogger(s.Log),
    )
    // In production, attach transport credentials with WithCreds (see TLS
    // below) or set GRPC_INSECURE=true; otherwise Build returns an error.

    // vel:grpc:services
    foo := services.NewFooService()
    p.server.RegisterService(func(srv interface{}) {
        foopb.RegisterFooServiceServer(srv.(*googleGrpc.Server), foo)
    })

    return nil
}

func (p *GRPCModule) Start(s *velocity.Services) error {
    if err := p.server.Build(); err != nil {
        return err
    }
    return p.server.StartAsync()
}

func (p *GRPCModule) Shutdown(ctx context.Context) error {
    return p.server.Shutdown(ctx)
}

The two marker comments are load-bearing: every later vel gen grpc service call injects the new proto import after // vel:grpc:imports and the new RegisterService block after // vel:grpc:services. Delete them and the generator prints a manual wire snippet instead of touching the file.

Register the module alongside the rest of the app’s modules:

v.Modules(func(r *velocity.ModuleRegistry) {
    r.Add(&modules.GRPCModule{})
})

velocity.New(velocity.WithModules(&modules.GRPCModule{})) works too; the gRPC module contributes no routes, listeners, or jobs, so it does not need the bootstrap-chain auto-wiring interfaces. See Modules for the difference between the two entry points.

Init binds the server and its service implementations; Start runs after every module has initialised. Build applies interceptors and registered services to the underlying *google.golang.org/grpc.Server; StartAsync launches the listener in a background goroutine so the rest of the app boot continues. Shutdown runs in reverse registration order and drains in-flight RPCs with the supplied ctx deadline.

Generators

Three generators produce the layout above in one call: proto, impl, module wiring, then buf generate.

vel gen grpc service

vel gen grpc service <Name> [flags]

Scaffolds a gRPC service end-to-end in one call:

• api/proto/<leaf>/v1/<name>.proto - empty service block
• api/proto/buf.yaml + api/proto/buf.gen.yaml (first run only)
• internal/grpc/services/<name>.go - <Name>Service impl with a New<Name>Service() constructor and the <alias>.Unimplemented<Name>ServiceServer embed
• internal/modules/grpc_module.go - created on first call (unless --no-module), then injected at // vel:grpc:imports and // vel:grpc:services markers on every subsequent call. The module wires the service via velgrpc.NewServer(...) and Register<Name>ServiceServer(...).

Name normalisation: vel gen grpc service Foo, FooService, foo, and fooService all produce the Go type FooService with proto package foo.v1 and default import alias foopb. The proto file uses option go_package = "<module>/api/gen/go/<leaf>/v1;<leaf>v1" derived from the host project’s go.mod (so the generated package itself is named <leaf>v1, referenced through the foopb alias).

buf.yaml / buf.gen.yaml are written before the proto file, so a config-write failure leaves no partial scaffold on disk. The generated GRPCModule does not hard-code WithReflection(true); it reads GRPC_PORT (default 50051) and otherwise takes the framework defaults, including reflection off unless GRPC_REFLECTION=true.

Wiring guards:

• If internal/modules/grpc_module.go already exists without the marker comments (legacy hand-written module), the command prints a manual wire snippet instead of mutating user code.
• If the existing module imports a services package other than this service’s impl directory, the command stops before writing any file and tells you to re-run with --no-module and wire it by hand.
• When the module already imports the generated proto package under a different alias, that alias is reused rather than emitting a duplicate import.

vel gen grpc service Foo
vel gen grpc service ChatService
vel gen grpc service TemplateControl --package admin \
  --proto-package velship.admin.v1 --dir internal/shared/grpc/services --no-module

After scaffolding, register &modules.GRPCModule{} in internal/app/bootstrap.go (printed as a hint on first run).

vel gen grpc rpc

vel gen grpc rpc <Service> <RPC> [--stream | --client-stream | --bidi]

Appends a new rpc to an existing service’s .proto and a matching method stub on the Go impl. The service must already exist; run vel gen grpc service <Name> first.

Only one streaming flag may be set per invocation; combining them errors out.

The proto scanner walks the file with brace counting that respects // line comments, /* block */ comments, and "..." string literals at every position (header keyword, between keyword and name, between name and {, and inside the body). That means rpc-with-options blocks (grpc-gateway HTTP annotations) and commented-out draft headers do not corrupt insertion.

On the Go side, the generated method signature matches the RPC shape (for service Foo, impl type FooService, default proto alias foopb):

context is added to the impl’s imports for unary only; streaming variants pull ctx from stream.Context() and do not need the import.

Idempotent: re-running with the same <Service> <RPC> pair detects the existing rpc and skips.

vel gen grpc rpc Foo Hello
vel gen grpc rpc Foo Tail --stream
vel gen grpc rpc Foo Upload --client-stream
vel gen grpc rpc Foo Chat --bidi

vel gen grpc gen

vel gen grpc gen

Runs buf generate inside api/proto. Streams buf’s stdout and stderr to your terminal so plugin errors are visible in real time. Takes no arguments, and fails with a clear message when:

• api/proto/ does not exist (run vel gen grpc service <Name> first)
• buf is not on PATH (links to install docs)
• buf generate exits non-zero

vel gen grpc gen
# cd api/proto && buf generate
# Generated Go code in api/gen/go/

Server

srv := grpc.NewServer(
    grpc.WithPort("50051"),
    grpc.WithEnvironment("production"),   // production guards: TLS required, reflection refused
    grpc.WithReflection(true),            // enable reflection in dev (Build errors in production)
    grpc.WithMaxRecvMsgSize(4<<20),
    grpc.WithMaxSendMsgSize(4<<20),
    grpc.WithCreds(creds),                // transport credentials (see TLS below)
    grpc.WithLogger(v.Log),
)

NewServer seeds its defaults from the environment via LoadConfig - GRPC_PORT, GRPC_REFLECTION, GRPC_MAX_RECV_SIZE, GRPC_MAX_SEND_SIZE

• and the explicit options above override them. WithMaxRecvMsgSize / WithMaxSendMsgSize clamp their argument: a non-positive size (which grpc-go reads as unlimited) falls back to the 4 MB default, and an oversize value is clamped to a 1 GiB ceiling.

TLS

Build enforces a production TLS guard. When the environment is production (APP_ENV=production or WithEnvironment("production")) and no transport credentials were attached, Build returns an error:

srv := grpc.NewServer(
    grpc.WithEnvironment("production"),
    grpc.WithCreds(creds),  // credentials.TransportCredentials
)

• WithCreds(creds) attaches credentials (from credentials.NewTLS, credentials.NewServerTLSFromFile, …) and marks the server as TLS-configured in one step.
• WithServerOption(grpc.Creds(...)) routes credentials through the raw option hook, but the guard cannot inspect it - pair it with WithExplicitTLS() so the guard recognises the opt-in.
• Setting GRPC_INSECURE=true opts a deployment out of the guard for a known-internal mTLS mesh or sidecar-terminated mesh.

Outside production, a missing credentials configuration only logs a one-shot warning.

Default recovery

Build installs a panic-recovery interceptor outermost by default (grpc-go does not auto-recover interceptor or handler panics, so without it the first panic crashes the serve loop). Pass WithoutDefaultRecovery() only when you wire your own outermost recovery interceptor first in the chain.

Interceptors

Unary and stream interceptors follow standard gRPC signatures:

srv.Use(
    authInterceptor,
    loggingInterceptor,
    recoveryInterceptor,
)

srv.UseStream(streamAuthInterceptor)

Mixed pairs:

srv.UseAll(
    grpc.InterceptorPair{Unary: authUnary, Stream: authStream},
)

Bundled interceptors

grpc/interceptors ships paired (unary + stream) implementations for the common middleware spine. Each returns an InterceptorPair, so UseAll wires both sides:

import (
    "github.com/velocitykode/velocity/grpc/interceptors"
    "github.com/velocitykode/velocity/grpc/grpcevents"
)

srv.UseAll(
    interceptors.Recovery(
        interceptors.WithRecoveryLogger(s.Log),
        interceptors.WithStackTrace(true),
        interceptors.WithRecoveryEventDispatcher(eventDispatcher),
    ),
    interceptors.Logging(
        interceptors.WithLoggingLogger(s.Log),
        interceptors.WithSlowThreshold(500*time.Millisecond),
        interceptors.WithSkipHealthChecks(true),
        interceptors.WithEventDispatcher(eventDispatcher),
    ),
    interceptors.Auth(validator,
        interceptors.WithPublicMethods("/foo.v1.FooService/Hello"),
        interceptors.WithAuthEventDispatcher(eventDispatcher),
    ),
)

validator implements interceptors.AuthValidator: given a token, return the resolved Claims or an error. WithPublicMethods bypasses auth on listed full-method paths. WithTokenExtractor overrides the default authorization: Bearer ... parser.

Each constructor accepts WithEventDispatcher (and variants) to wire the matching grpcevents event onto a domain bus. The dispatchers also start or continue a trace span on entry, so RequestStarted / RequestCompleted / RequestFailed ship TraceID / SpanID / ParentID (the standard 3-field convention).

Registering services

srv.RegisterService(func(s any) {
    pb.RegisterUsersServer(s.(*googleGrpc.Server), usersImpl)
})

The closure receives the underlying *grpc.Server - use it to call generated Register* functions.

Running

if err := srv.Build(); err != nil {
    return err
}

if err := srv.Start(); err != nil {  // blocks
    return err
}

Non-blocking:

if err := srv.StartAsync(); err != nil {
    return err
}

// later
srv.GracefulStop()

Shutdown(ctx) waits for in-flight RPCs with a deadline; Stop is immediate.

RPC types

gRPC has four method shapes. The framework treats all of them uniformly: register the service with srv.RegisterService, the bundled interceptors trace both unary and stream RPCs via paired Unary + Stream implementations.

Three things change per shape, one thing stays the same:

Same proto file declares all four with the stream keyword in the right position:

service FooService {
  rpc Hello(HelloRequest)        returns (HelloResponse);          // unary
  rpc Tail(HelloRequest)         returns (stream HelloResponse);   // server stream
  rpc Upload(stream HelloRequest) returns (HelloResponse);         // client stream
  rpc Chat(stream HelloRequest)  returns (stream HelloResponse);   // bidi
}

buf generate produces matching server-side handler signatures; the service impl below has one method per RPC, each with the signature shape its row prescribes. Registration is unchanged:

foopb.RegisterFooServiceServer(srv, fooImpl) // same call regardless of mix

Unary

One request, one response. Standard request/response RPC.

Use cases:

• CRUD endpoints (GetUser, CreateOrder, UpdateProfile)
• Auth handshake (Login, RefreshToken)
• Internal service-to-service calls (ChargeCard, SendReceipt)
• Synchronous validation / pricing / risk-score lookups
• Anything that fits a REST GET / POST mental model

service ChatService {
  rpc SendMessage(SendMessageRequest) returns (SendMessageResponse);
}

func (s *ChatService) SendMessage(ctx context.Context, req *pb.SendMessageRequest) (*pb.SendMessageResponse, error) {
    msg, err := s.messages.Create(ctx, req.GetBody())
    if err != nil {
        return nil, velgrpc.WrapError(err)
    }
    return &pb.SendMessageResponse{Id: msg.ID}, nil
}

Server streaming

One request, many responses.

Use cases:

• Log / metric tailing (TailLogs(filter) streams new lines as they arrive)
• Server-pushed notifications (Subscribe(topic) for chat, alerts, presence)
• Large result sets streamed in chunks (Export(query) without materialising the full set in memory)
• Progress updates for long-running jobs (RunBuild(jobID) emits stage events)
• Server-Sent-Events replacement on the gRPC wire

service ChatService {
  rpc Subscribe(SubscribeRequest) returns (stream Message);
}

func (s *ChatService) Subscribe(req *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error {
    ctx := stream.Context()
    sub, err := s.pubsub.Subscribe(ctx, req.GetTopic())
    if err != nil {
        return velgrpc.WrapError(err)
    }
    defer sub.Close()

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case msg, ok := <-sub.Ch():
            if !ok {
                return nil
            }
            if err := stream.Send(&pb.Message{Body: msg.Body}); err != nil {
                return err
            }
        }
    }
}

Honor stream.Context().Done(): the client may cancel mid-stream, and the framework’s recovery interceptor will not save a handler that ignores cancellation.

Client streaming

Many requests, one response.

Use cases:

• Bulk ingestion (UploadEvents, UploadMetrics, UploadRows)
• File upload chunked over the wire (PutObject streams blob chunks, returns a single ack with the resolved object id)
• Telemetry batching from edge agents into a central collector
• Aggregation / reduce APIs (ComputeStats consumes many samples, returns one summary)
• Anything the client can produce incrementally but the server only needs to acknowledge once at the end

service IngestService {
  rpc UploadEvents(stream Event) returns (UploadEventsResponse);
}

func (s *IngestService) UploadEvents(stream pb.IngestService_UploadEventsServer) error {
    var count int64
    for {
        evt, err := stream.Recv()
        if err == io.EOF {
            return stream.SendAndClose(&pb.UploadEventsResponse{Accepted: count})
        }
        if err != nil {
            return err
        }
        if err := s.store.Append(stream.Context(), evt); err != nil {
            return velgrpc.WrapError(err)
        }
        count++
    }
}

stream.SendAndClose ends the RPC on the success path. Returning an error short-circuits the stream with that status.

Bidirectional streaming

Many requests, many responses, fully interleaved.

Use cases:

• Chat / DMs / presence (Connect() carries inbound messages and outbound deliveries on one long-lived stream)
• Interactive shells / remote exec (Exec() ships stdin chunks, returns stdout/stderr chunks)
• Live collaboration (CRDT / OT sync, multi-cursor editors)
• Pub/sub bridges where the client both publishes and subscribes on the same connection
• Real-time game / robotics control loops (client sends inputs, server sends state updates at its own cadence)

service ChatService {
  rpc Connect(stream ClientFrame) returns (stream ServerFrame);
}

func (s *ChatService) Connect(stream pb.ChatService_ConnectServer) error {
    ctx := stream.Context()

    // Outbound pump: server -> client
    outbound := make(chan *pb.ServerFrame, 16)
    go func() {
        defer close(outbound)
        s.fanIn(ctx, outbound) // implementation-specific
    }()

    // Inbound pump runs on the request goroutine; send loop pulls from `outbound`.
    inboundErr := make(chan error, 1)
    go func() {
        for {
            frame, err := stream.Recv()
            if err != nil {
                inboundErr <- err
                return
            }
            if err := s.handle(ctx, frame); err != nil {
                inboundErr <- err
                return
            }
        }
    }()

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case err := <-inboundErr:
            if err == io.EOF {
                return nil
            }
            return err
        case frame, ok := <-outbound:
            if !ok {
                return nil
            }
            if err := stream.Send(frame); err != nil {
                return err
            }
        }
    }
}

stream.Recv() and stream.Send() are independently safe to call from different goroutines, but Recv from two goroutines or Send from two goroutines is undefined; serialise each direction.

HTTP gateway (grpc-gateway)

Expose the same service over REST:

gw := grpc.NewGateway(
    grpc.GatewayWithPort("8080"),
    grpc.GatewayWithGRPCEndpoint("localhost:50051"),
    grpc.GatewayWithInsecure(),  // or GatewayWithTLS(certFile)
    grpc.GatewayWithLogger(v.Log),
)

gw.RegisterHandler(pb.RegisterUsersHandlerFromEndpoint)

if err := gw.Build(context.Background()); err != nil {
    return err
}

// HTTP middleware chains applied to the gateway mux
gw.Use(cors.Default().Handler, requestLogger)

gw.StartAsync()

gw.Mux() returns the underlying runtime.ServeMux if you need to attach custom handlers alongside the generated ones.

Like the server, the gateway enforces a production TLS guard: in production it refuses to start unless dial credentials were configured via GatewayWithTLS(caCertFile), GatewayWithTransportConfig(cfg) (typed mTLS config with TLSCert / TLSKey / CACert), or GatewayWithInsecure() to opt out for a known-internal mesh. Outside production an unconfigured gateway defaults to insecure credentials and logs a one-shot warning. The gateway’s HTTP server ships conservative read / write / idle timeouts and a header-size bound by default; override them with GatewayWithReadTimeout, GatewayWithWriteTimeout, GatewayWithIdleTimeout, and GatewayWithMaxHeaderBytes. Defaults for GatewayWithPort / GatewayWithGRPCEndpoint come from GATEWAY_PORT / GRPC_ENDPOINT.

Context helpers

Claims

Claims is an interface (GetUserID() uint / GetTeamID() uint); apps implement it with their own claims type, or use the bundled BasicClaims:

ctx = grpc.ContextWithClaims(ctx, &grpc.BasicClaims{UserID: 42, TeamID: 7})

claims := grpc.ClaimsFromContext(ctx)       // returns nil if absent
claims := grpc.MustClaimsFromContext(ctx)   // panics if absent - use in auth'd handlers

userID := grpc.UserIDFromContext(ctx)       // shortcut, 0 if no claims
teamID := grpc.TeamIDFromContext(ctx)

Attach them in an auth interceptor before the handler runs. The bundled interceptors.Auth does this for you: its AuthValidator.ValidateToken returns the resolved Claims, which the interceptor stores on the context.

Request ID

ctx = grpc.ContextWithRequestID(ctx, grpc.GenerateRequestID())
id := grpc.RequestIDFromContext(ctx)

Method name

ctx = grpc.ContextWithMethod(ctx, info.FullMethod)
method := grpc.MethodFromContext(ctx)

Useful inside logging interceptors that run below the method dispatch.

Metadata helpers

token  := grpc.ExtractBearerToken(ctx)       // from "authorization: Bearer ..."
value  := grpc.ExtractMetadata(ctx, "x-tenant-id")
values := grpc.ExtractAllMetadata(ctx, "x-forwarded-for")

Errors

Typed constructors produce standard google.golang.org/grpc/status errors:

return nil, grpc.NotFound("user not found")
return nil, grpc.NotFoundf("user %d", id)
return nil, grpc.InvalidArgument("id required")
return nil, grpc.Unauthenticated("missing token")
return nil, grpc.PermissionDenied("not your team")
return nil, grpc.AlreadyExists("duplicate email")
return nil, grpc.FailedPrecondition("invoice already finalized")
return nil, grpc.ResourceExhausted("quota exceeded")
return nil, grpc.Unavailable("downstream timeout")
return nil, grpc.Internal("bug")

All wrap codes.Code under the hood.

Inspecting errors

if grpc.IsNotFound(err) {
    // handle
}

switch grpc.Code(err) {
case codes.NotFound:         // ...
case codes.InvalidArgument:  // ...
}

msg := grpc.Message(err)        // unwrap status message
status := grpc.FromError(err)   // raw *status.Status

Wrapping existing errors

if err := db.Find(&user); err != nil {
    return nil, grpc.WrapError(err)  // maps stdlib errors → grpc codes
}

return nil, grpc.WrapErrorWithCode(err, codes.FailedPrecondition)

Reflection in production

When the environment is production (APP_ENV=production or WithEnvironment("production")), Build hard-fails if reflection is enabled - it returns an error rather than silently downgrading, so a misconfigured deployment cannot ship thinking reflection is on while it is off (or vice versa). Disable it (GRPC_REFLECTION=false or drop WithReflection(true)) for production builds. "prod" and "staging" fold into the same locked-down branch, so a typo’d APP_ENV cannot re-enable introspection.

Health checks

The package ships a standard gRPC health service. Construct one with grpc.NewHealthService() and register it via the server’s RegisterService:

health := grpc.NewHealthService()
srv.RegisterService(health.RegistrationFunc())

Drive the reported status with health.SetServing(service) / health.SetNotServing(service), or register a dynamic probe with health.RegisterChecker(service, func(ctx) error { ... }) that runs on each check. Pair it with your load balancer’s gRPC health probes.

Events

grpc/grpcevents defines the events the bundled interceptors and the server lifecycle dispatch. Wire a dispatcher via Server.SetEventDispatcher (or pass the matching With*EventDispatcher option into each bundled interceptor) and route them onto the framework event bus:

srv.SetEventDispatcher(func(ctx context.Context, evt any) error {
    return s.Events.Dispatch(ctx, evt)
})

Protocol (“grpc” or “http”) on each request/stream event identifies whether the call entered through the native gRPC server or the HTTP gateway.

Converters

grpc/converters ships helpers for the two most common proto <-> Go gaps:

import "github.com/velocitykode/velocity/grpc/converters"

// time.Time <-> *timestamppb.Timestamp
ts := converters.TimeValueToProto(user.CreatedAt)
t  := converters.ProtoToTimeValue(req.GetCreatedAt())

// time.Duration <-> *durationpb.Duration
d  := converters.DurationToProto(15 * time.Minute)
dur := converters.ProtoToDuration(req.GetTimeout())

// Offset pagination
page := converters.NormalizePagination(req.GetPage(), req.GetPageSize())
resp := converters.NewPaginationResponse(int32(page.Page), int32(page.Size), totalItems)

// Cursor pagination
cp := converters.NormalizeCursorPagination(req.GetCursor(), req.GetLimit())
cr := converters.NewCursorResponse(nextCursor, prevCursor, hasMore, int(cp.Limit))

NormalizePagination clamps page / page-size to sane defaults so handlers don’t have to defensively validate. CalculateTotalPages and the cursor helpers cover the remaining bookkeeping.