Golang gRPC Microservices: Protobuf, TLS & Middleware

// proto/driver/v1/driver.proto
syntax = "proto3";

package driver.v1;

option go_package = "github.com/yourorg/platform/gen/driver/v1;driverv1";

import "google/protobuf/timestamp.proto";

// DriverService manages driver location and availability
service DriverService {
  // Unary: Get a single driver by ID
  rpc GetDriver(GetDriverRequest) returns (GetDriverResponse);

  // Server streaming: Track driver location in real time
  rpc StreamLocation(StreamLocationRequest) returns (stream LocationUpdate);

  // Client streaming: Driver app sends bulk GPS updates
  rpc UploadLocations(stream LocationUpdate) returns (UploadSummary);

  // Bidirectional: Full-duplex driver-server communication
  rpc DriverSession(stream DriverEvent) returns (stream ServerCommand);
}

message GetDriverRequest {
  string driver_id = 1;
}

message GetDriverResponse {
  string driver_id = 1;
  string status = 2;          // AVAILABLE, BUSY, OFFLINE
  double latitude = 3;
  double longitude = 4;
  google.protobuf.Timestamp last_seen_at = 5;
}

message StreamLocationRequest {
  string driver_id = 1;
}

message LocationUpdate {
  string driver_id = 1;
  double latitude = 2;
  double longitude = 3;
  float speed_mps = 4;
  float heading_degrees = 5;
  google.protobuf.Timestamp timestamp = 6;
}

message UploadSummary {
  int32 received_count = 1;
  int32 persisted_count = 2;
  string session_id = 3;
}

message DriverEvent {
  oneof event {
    LocationUpdate location = 1;
    DriverStatusChange status_change = 2;
    HeartbeatPing heartbeat = 3;
  }
}

message ServerCommand {
  oneof command {
    RideOffer ride_offer = 1;
    NavigationUpdate navigation = 2;
    PingResponse pong = 3;
  }
}

message DriverStatusChange {
  string driver_id = 1;
  string new_status = 2;
}

message HeartbeatPing { int64 client_ts_ms = 1; }
message PingResponse { int64 server_ts_ms = 1; }
message RideOffer { string offer_id = 1; string pickup_address = 2; }
message NavigationUpdate { string polyline = 1; }

# Install tools
go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest

# Generate — run from project root
protoc \
  --go_out=gen \
  --go_opt=paths=source_relative \
  --go-grpc_out=gen \
  --go-grpc_opt=paths=source_relative \
  proto/driver/v1/driver.proto

// internal/driver/server.go
package driver

import (
    "context"
    "fmt"
    "io"
    "log/slog"
    "time"

    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"
    "google.golang.org/protobuf/types/known/timestamppb"

    driverv1 "github.com/yourorg/platform/gen/driver/v1"
)

// Server implements driverv1.DriverServiceServer
type Server struct {
    driverv1.UnimplementedDriverServiceServer
    repo      DriverRepository
    publisher LocationPublisher
    logger    *slog.Logger
}

func NewServer(repo DriverRepository, pub LocationPublisher, log *slog.Logger) *Server {
    return &Server{repo: repo, publisher: pub, logger: log}
}

// GetDriver — Unary RPC
func (s *Server) GetDriver(ctx context.Context, req *driverv1.GetDriverRequest) (*driverv1.GetDriverResponse, error) {
    if req.DriverId == "" {
        return nil, status.Error(codes.InvalidArgument, "driver_id is required")
    }

    driver, err := s.repo.FindByID(ctx, req.DriverId)
    if err != nil {
        s.logger.ErrorContext(ctx, "GetDriver: repo error", "driver_id", req.DriverId, "err", err)
        return nil, status.Errorf(codes.Internal, "failed to fetch driver: %v", err)
    }
    if driver == nil {
        return nil, status.Errorf(codes.NotFound, "driver %s not found", req.DriverId)
    }

    return &driverv1.GetDriverResponse{
        DriverId:   driver.ID,
        Status:     driver.Status,
        Latitude:   driver.Lat,
        Longitude:  driver.Lng,
        LastSeenAt: timestamppb.New(driver.LastSeenAt),
    }, nil
}

// StreamLocation — Server-streaming RPC
// Sends the driver's live location to the caller every 2 seconds
func (s *Server) StreamLocation(req *driverv1.StreamLocationRequest, stream driverv1.DriverService_StreamLocationServer) error {
    ctx := stream.Context()

    for {
        select {
        case <-ctx.Done():
            return nil // Client disconnected
        case <-time.After(2 * time.Second):
            loc, err := s.repo.GetCurrentLocation(ctx, req.DriverId)
            if err != nil {
                return status.Errorf(codes.Internal, "location fetch failed: %v", err)
            }
            if err := stream.Send(&driverv1.LocationUpdate{
                DriverId:  req.DriverId,
                Latitude:  loc.Lat,
                Longitude: loc.Lng,
                Timestamp: timestamppb.Now(),
            }); err != nil {
                return err // Client disconnected mid-stream
            }
        }
    }
}

// UploadLocations — Client-streaming RPC
// Driver app uploads batched GPS points; server aggregates and persists
func (s *Server) UploadLocations(stream driverv1.DriverService_UploadLocationsServer) error {
    var received, persisted int32
    var sessionID string

    for {
        update, err := stream.Recv()
        if err == io.EOF {
            // Client finished sending; send summary response
            return stream.SendAndClose(&driverv1.UploadSummary{
                ReceivedCount: received,
                PersistedCount: persisted,
                SessionId:     sessionID,
            })
        }
        if err != nil {
            return status.Errorf(codes.Internal, "recv error: %v", err)
        }

        received++
        sessionID = fmt.Sprintf("sess-%s-%d", update.DriverId, time.Now().UnixMilli())

        if err := s.publisher.Publish(stream.Context(), update); err != nil {
            s.logger.Warn("publish failed", "driver_id", update.DriverId, "err", err)
            continue // Skip failed publishes, don't abort the whole batch
        }
        persisted++
    }
}

// DriverSession — Bidirectional streaming RPC
func (s *Server) DriverSession(stream driverv1.DriverService_DriverSessionServer) error {
    ctx := stream.Context()

    for {
        event, err := stream.Recv()
        if err == io.EOF {
            return nil
        }
        if err != nil {
            return err
        }

        switch e := event.Event.(type) {
        case *driverv1.DriverEvent_Location:
            _ = s.publisher.Publish(ctx, e.Location)

        case *driverv1.DriverEvent_Heartbeat:
            if err := stream.Send(&driverv1.ServerCommand{
                Command: &driverv1.ServerCommand_Pong{
                    Pong: &driverv1.PingResponse{ServerTsMs: time.Now().UnixMilli()},
                },
            }); err != nil {
                return err
            }

        case *driverv1.DriverEvent_StatusChange:
            s.logger.InfoContext(ctx, "driver status changed",
                "driver_id", e.StatusChange.DriverId,
                "new_status", e.StatusChange.NewStatus,
            )
        }
    }
}

// internal/interceptor/chain.go
package interceptor

import (
    "context"
    "log/slog"
    "runtime/debug"
    "time"

    "google.golang.org/grpc"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/metadata"
    "google.golang.org/grpc/status"
)

// LoggingUnaryInterceptor logs method, duration, and status code for every RPC.
func LoggingUnaryInterceptor(logger *slog.Logger) grpc.UnaryServerInterceptor {
    return func(
        ctx context.Context,
        req any,
        info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler,
    ) (any, error) {
        start := time.Now()
        resp, err := handler(ctx, req)

        code := codes.OK
        if err != nil {
            code = status.Code(err)
        }

        logger.InfoContext(ctx, "grpc unary",
            "method", info.FullMethod,
            "duration_ms", time.Since(start).Milliseconds(),
            "code", code.String(),
        )
        return resp, err
    }
}

// AuthUnaryInterceptor validates the Authorization header.
func AuthUnaryInterceptor(tokenValidator TokenValidator) grpc.UnaryServerInterceptor {
    return func(
        ctx context.Context,
        req any,
        info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler,
    ) (any, error) {
        md, ok := metadata.FromIncomingContext(ctx)
        if !ok {
            return nil, status.Error(codes.Unauthenticated, "missing metadata")
        }

        tokens := md.Get("authorization")
        if len(tokens) == 0 {
            return nil, status.Error(codes.Unauthenticated, "missing authorization token")
        }

        claims, err := tokenValidator.Validate(tokens[0])
        if err != nil {
            return nil, status.Errorf(codes.Unauthenticated, "invalid token: %v", err)
        }

        // Inject claims into context for downstream handlers
        ctx = context.WithValue(ctx, claimsKey{}, claims)
        return handler(ctx, req)
    }
}

// RecoveryUnaryInterceptor catches panics and converts them to gRPC Internal errors.
func RecoveryUnaryInterceptor(logger *slog.Logger) grpc.UnaryServerInterceptor {
    return func(
        ctx context.Context,
        req any,
        info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler,
    ) (resp any, err error) {
        defer func() {
            if r := recover(); r != nil {
                logger.ErrorContext(ctx, "panic recovered",
                    "method", info.FullMethod,
                    "panic", r,
                    "stack", string(debug.Stack()),
                )
                err = status.Errorf(codes.Internal, "internal server error")
            }
        }()
        return handler(ctx, req)
    }
}

type claimsKey struct{}
type TokenValidator interface {
    Validate(token string) (Claims, error)
}
type Claims struct{ SubjectID string }

// cmd/server/main.go
package main

import (
    "crypto/tls"
    "crypto/x509"
    "fmt"
    "log"
    "net"
    "os"
    "os/signal"
    "syscall"
    "time"

    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials"
    "google.golang.org/grpc/health"
    "google.golang.org/grpc/health/grpc_health_v1"
    "google.golang.org/grpc/keepalive"
    "google.golang.org/grpc/reflection"

    driverv1 "github.com/yourorg/platform/gen/driver/v1"
    "github.com/yourorg/platform/internal/driver"
    "github.com/yourorg/platform/internal/interceptor"
)

func main() {
    // --- mTLS credentials ---
    cert, err := tls.LoadX509KeyPair("certs/server.crt", "certs/server.key")
    if err != nil {
        log.Fatalf("load server cert: %v", err)
    }

    caCert, err := os.ReadFile("certs/ca.crt")
    if err != nil {
        log.Fatalf("read CA cert: %v", err)
    }
    caPool := x509.NewCertPool()
    caPool.AppendCertsFromPEM(caCert)

    tlsCreds := credentials.NewTLS(&tls.Config{
        Certificates: []tls.Certificate{cert},
        ClientAuth:   tls.RequireAndVerifyClientCert, // mTLS: require client cert
        ClientCAs:    caPool,
        MinVersion:   tls.VersionTLS13,
    })

    // --- Build gRPC server with interceptor chain ---
    logger := slog.New(slog.NewJSONHandler(os.Stdout, nil))

    srv := grpc.NewServer(
        grpc.Creds(tlsCreds),
        grpc.ChainUnaryInterceptor(
            interceptor.RecoveryUnaryInterceptor(logger),   // Must be first — catches panics from all others
            interceptor.LoggingUnaryInterceptor(logger),
            interceptor.AuthUnaryInterceptor(tokenValidator),
        ),
        // Keepalive: prevent silent connection drops behind NAT/load balancers
        grpc.KeepaliveParams(keepalive.ServerParameters{
            MaxConnectionIdle:     15 * time.Minute,
            MaxConnectionAge:      30 * time.Minute,
            MaxConnectionAgeGrace: 5 * time.Second,
            Time:                  5 * time.Minute,
            Timeout:               1 * time.Second,
        }),
        grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
            MinTime:             5 * time.Second,
            PermitWithoutStream: true,
        }),
    )

    // --- Register services ---
    driverServer := driver.NewServer(repo, publisher, logger)
    driverv1.RegisterDriverServiceServer(srv, driverServer)

    // Health check — required by Kubernetes liveness probes and gRPC load balancers
    healthSrv := health.NewServer()
    grpc_health_v1.RegisterHealthServer(srv, healthSrv)
    healthSrv.SetServingStatus("driver.v1.DriverService", grpc_health_v1.HealthCheckResponse_SERVING)

    // Reflection — enables grpcurl and Postman gRPC without importing .proto files
    reflection.Register(srv)

    // --- Start listening ---
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        log.Fatalf("listen: %v", err)
    }

    log.Printf("gRPC server listening on :50051")

    // --- Graceful shutdown ---
    go func() {
        if err := srv.Serve(lis); err != nil {
            log.Printf("serve error: %v", err)
        }
    }()

    quit := make(chan os.Signal, 1)
    signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
    <-quit

    log.Println("shutting down gRPC server...")
    healthSrv.SetServingStatus("driver.v1.DriverService", grpc_health_v1.HealthCheckResponse_NOT_SERVING)
    srv.GracefulStop() // Waits for in-flight RPCs to finish
    log.Println("server stopped")
}

// internal/client/driver_client.go
package client

import (
    "context"
    "crypto/tls"
    "crypto/x509"
    "log"
    "os"
    "time"

    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials"
    "google.golang.org/grpc/keepalive"

    driverv1 "github.com/yourorg/platform/gen/driver/v1"
)

func NewDriverClient(target string) (driverv1.DriverServiceClient, func(), error) {
    // mTLS client credentials
    cert, err := tls.LoadX509KeyPair("certs/client.crt", "certs/client.key")
    if err != nil {
        return nil, nil, fmt.Errorf("load client cert: %w", err)
    }

    caCert, _ := os.ReadFile("certs/ca.crt")
    caPool := x509.NewCertPool()
    caPool.AppendCertsFromPEM(caCert)

    creds := credentials.NewTLS(&tls.Config{
        Certificates: []tls.Certificate{cert},
        RootCAs:      caPool,
        MinVersion:   tls.VersionTLS13,
    })

    conn, err := grpc.NewClient(
        target,
        grpc.WithTransportCredentials(creds),
        // Default round-robin load balancing across multiple server instances
        grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`),
        grpc.WithKeepaliveParams(keepalive.ClientParameters{
            Time:                10 * time.Minute,
            Timeout:             5 * time.Second,
            PermitWithoutStream: true,
        }),
    )
    if err != nil {
        return nil, nil, fmt.Errorf("dial %s: %w", target, err)
    }

    cleanup := func() { conn.Close() }
    return driverv1.NewDriverServiceClient(conn), cleanup, nil
}

// Usage example
func exampleGetDriver(ctx context.Context) {
    client, cleanup, err := NewDriverClient("dns:///driver-service:50051")
    if err != nil {
        log.Fatal(err)
    }
    defer cleanup()

    ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
    defer cancel()

    resp, err := client.GetDriver(ctx, &driverv1.GetDriverRequest{DriverId: "drv-abc123"})
    if err != nil {
        log.Printf("GetDriver error: %v", err)
        return
    }
    log.Printf("Driver %s is %s at (%f, %f)", resp.DriverId, resp.Status, resp.Latitude, resp.Longitude)
}

# Dockerfile — multi-stage build for minimal image size
FROM golang:1.23-alpine AS builder
WORKDIR /app

COPY go.mod go.sum ./
RUN go mod download

COPY . .
RUN CGO_ENABLED=0 GOOS=linux go build -o /bin/driver-service ./cmd/server

FROM gcr.io/distroless/static-debian12
COPY --from=builder /bin/driver-service /driver-service
COPY certs/ /certs/

EXPOSE 50051
ENTRYPOINT ["/driver-service"]

# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: driver-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: driver-service
  template:
    metadata:
      labels:
        app: driver-service
    spec:
      containers:
        - name: driver-service
          image: yourorg/driver-service:latest
          ports:
            - containerPort: 50051
              name: grpc
          livenessProbe:
            grpc:
              port: 50051
              service: driver.v1.DriverService
            initialDelaySeconds: 10
            periodSeconds: 10
          readinessProbe:
            grpc:
              port: 50051
              service: driver.v1.DriverService
            initialDelaySeconds: 5
            periodSeconds: 5
          resources:
            requests:
              cpu: "100m"
              memory: "128Mi"
            limits:
              cpu: "500m"
              memory: "512Mi"

// ❌ Bad: No deadline — if the server hangs, the goroutine leaks forever
resp, err := client.GetDriver(context.Background(), req)

// ✅ Good: Always set a deadline
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
resp, err := client.GetDriver(ctx, req)

// ❌ Bad: Treating all errors the same
if err != nil {
    return fmt.Errorf("grpc error: %v", err)
}

// ✅ Good: Check the status code for retryability
if err != nil {
    st, _ := status.FromError(err)
    switch st.Code() {
    case codes.NotFound:
        return nil, ErrDriverNotFound
    case codes.Unavailable, codes.ResourceExhausted:
        // Retryable — apply backoff
        return nil, ErrRetryable
    default:
        return nil, err
    }
}

// Without keepalive, NAT firewalls silently drop idle gRPC streams after ~4 minutes.
// Result: the client thinks it's connected but receives no messages.
// Fix: configure keepalive on both client and server (shown in Step 4 and 5 above).

// ❌ Bad: gRPC default is pick_first — all traffic goes to one pod
conn, _ := grpc.NewClient("dns:///driver-service:50051", grpc.WithTransportCredentials(creds))

// ✅ Good: round_robin distributes across all healthy pods
conn, _ := grpc.NewClient(
    "dns:///driver-service:50051",
    grpc.WithTransportCredentials(creds),
    grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`),
)