Golang gRPC Microservices chuẩn Production

// 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";

// Dịch vụ DriverService quản trị vị trí và sự hiện diện của lái xe
service DriverService {
  // Lệnh Unary: Trích xuất một tài xế đơn lẻ bằng mã ID
  rpc GetDriver(GetDriverRequest) returns (GetDriverResponse);

  // Streaming từ chiều Server (Server streaming): Dò bắt địa điểm xe chạy ở khung thời gian thực
  rpc StreamLocation(StreamLocationRequest) returns (stream LocationUpdate);

  // Streaming từ chiều Client (Client streaming): App tài xế truyền gộp cục (bulk) liên tiếp GPS updates
  rpc UploadLocations(stream LocationUpdate) returns (UploadSummary);

  // Streaming hai chiều (Bidirectional): Liên lạc truyền dẫn tín hiệu song công (Full-duplex) giữa tài xế-máy chủ
  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; }

# Mang cài mớ đồ nghề
go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest

# Phát động tiến trình render Code — ấn gọi lệnh tính từ vùng gốc của project
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"
)

// Dựng khung cho hệ Server triển khai phần lõi 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}
}

// Hàm GetDriver — Mẫu Unary RPC tiêu chuẩn
func (s *Server) GetDriver(ctx context.Context, req *driverv1.GetDriverRequest) (*driverv1.GetDriverResponse, error) {
    if req.DriverId == "" {
        return nil, status.Error(codes.InvalidArgument, "bắt buộc phải truyền mã driver_id")
    }

    driver, err := s.repo.FindByID(ctx, req.DriverId)
    if err != nil {
        s.logger.ErrorContext(ctx, "GetDriver: lỗi truy vấn repo error", "driver_id", req.DriverId, "err", err)
        return nil, status.Errorf(codes.Internal, "việc trích xuất tài xế gặp lỗi: %v", err)
    }
    if driver == nil {
        return nil, status.Errorf(codes.NotFound, "đã tìm nhưng không thấy dấu vết tài xế %s", req.DriverId)
    }

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

// Khối chức năng StreamLocation — Đây là loại Server-streaming RPC
// Nhiệm vụ là ném ra trả lại vị trí sống live location của tài xế vào mặt người truy xuất (caller) cứ độ nhịp 2 giây một phát
func (s *Server) StreamLocation(req *driverv1.StreamLocationRequest, stream driverv1.DriverService_StreamLocationServer) error {
    ctx := stream.Context()

    for {
        select {
        case <-ctx.Done():
            return nil // Client đã bấm ngắt kết nối
        case <-time.After(2 * time.Second):
            loc, err := s.repo.GetCurrentLocation(ctx, req.DriverId)
            if err != nil {
                return status.Errorf(codes.Internal, "chức năng kéo location gặp sự cố thất bại: %v", err)
            }
            if err := stream.Send(&driverv1.LocationUpdate{
                DriverId:  req.DriverId,
                Latitude:  loc.Lat,
                Longitude: loc.Lng,
                Timestamp: timestamppb.Now(),
            }); err != nil {
                return err // Phía Client tự ý cắt kết nối ngay giữa chừng cơn bão stream
            }
        }
    }
}

// Hệ thống UploadLocations — Mô hình Client-streaming RPC
// Phía app ứng dụng bác tài gửi đóng gói từng bó batched tọa độ GPS; máy chủ chịu trách nhiệm hứng tổng hợp lại rồi save lưu xuống kho
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 {
            // Phía Client đã hết hàng đẩy qua; đóng luồng gửi kết quả phản hồi summary
            return stream.SendAndClose(&driverv1.UploadSummary{
                ReceivedCount: received,
                PersistedCount: persisted,
                SessionId:     sessionID,
            })
        }
        if err != nil {
            return status.Errorf(codes.Internal, "đụng độ lỗi phần tiếp nhận 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("gặp lỗi khâu phát lệnh publish", "driver_id", update.DriverId, "err", err)
            continue // Cho ngó lơ bỏ qua mấy nhát hụt publish, đừng thấy khó mà ngưng ngang hông phá bĩnh nguyên cụm bó whole batch
        }
        persisted++
    }
}

// Hàm DriverSession — Dạng 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, "trạng thái bên lái xe chuyển đổi chớp nháy",
                "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 đi theo dõi và ghi chép lại các tên hàm, quãng thời lượng, và đoạn mã trạng thái trả về ứng cho từng cái 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, "dòng grpc unary",
            "method", info.FullMethod,
            "duration_ms", time.Since(start).Milliseconds(),
            "code", code.String(),
        )
        return resp, err
    }
}

// Hàm AuthUnaryInterceptor lo phần phân bua xác minh cái thẻ Header Authorization.
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, "kêu thiếu phần siêu dữ liệu metadata")
        }

        tokens := md.Get("authorization")
        if len(tokens) == 0 {
            return nil, status.Error(codes.Unauthenticated, "vắng bóng chuỗi mã authorization token")
        }

        claims, err := tokenValidator.Validate(tokens[0])
        if err != nil {
            return nil, status.Errorf(codes.Unauthenticated, "vật chứa mã token lậu: %v", err)
        }

        // Bơm chèn trực tiếp các claims nhét vào vô ngữ cảnh context gửi tuốt luốt cho mấy tay sai downstream handlers ở dưới
        ctx = context.WithValue(ctx, claimsKey{}, claims)
        return handler(ctx, req)
    }
}

// Chốt RecoveryUnaryInterceptor giăng lưới tóm bắt các vụ nổ panics và khéo léo bọc chuyển hóa chúng đè thành các bộ lỗi gRPC Internal.
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, "cứu giá và khôi phục thành công panic",
                    "method", info.FullMethod,
                    "panic", r,
                    "stack", string(debug.Stack()),
                )
                err = status.Errorf(codes.Internal, "lỗi nội vi cục bộ 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() {
    // --- Lắp mTLS credentials ---
    cert, err := tls.LoadX509KeyPair("certs/server.crt", "certs/server.key")
    if err != nil {
        log.Fatalf("chết bước tải server cert: %v", err)
    }

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

    tlsCreds := credentials.NewTLS(&tls.Config{
        Certificates: []tls.Certificate{cert},
        ClientAuth:   tls.RequireAndVerifyClientCert, // Chuẩn mTLS: bắt nộp bằng được cái client cert
        ClientCAs:    caPool,
        MinVersion:   tls.VersionTLS13,
    })

    // --- Khởi tạo khối gRPC server với cái băng chuyền đánh chặn interceptor chain ---
    logger := slog.New(slog.NewJSONHandler(os.Stdout, nil))

    srv := grpc.NewServer(
        grpc.Creds(tlsCreds),
        grpc.ChainUnaryInterceptor(
            interceptor.RecoveryUnaryInterceptor(logger),   // Thằng này BẮT BUỘC chễm chệ chiếm ngôi số 1 — lo vớt vát mớ nổ panics phang ra từ các đàn em rớt lại
            interceptor.LoggingUnaryInterceptor(logger),
            interceptor.AuthUnaryInterceptor(tokenValidator),
        ),
        // Chức năng Keepalive: phòng bị hệ thống lặng lẽ chết êm ái cắt nguồn kết nối khi thụt lùi ẩn núp đằng sau lưng các bộ cân 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,
        }),
    )

    // --- Thông quan các dịch vụ (Register services) ---
    driverServer := driver.NewServer(repo, publisher, logger)
    driverv1.RegisterDriverServiceServer(srv, driverServer)

    // Kiểm tra sức khỏe (Health check) — yếu tố sống còn đòi hỏi bởi bộ giám sát sinh tồn liveness probes ở bên vương quốc Kubernetes chung cùng với tụi gRPC load balancers
    healthSrv := health.NewServer()
    grpc_health_v1.RegisterHealthServer(srv, healthSrv)
    healthSrv.SetServingStatus("driver.v1.DriverService", grpc_health_v1.HealthCheckResponse_SERVING)

    // Nhãn quan (Reflection) — thứ kích hoạt chức năng xịn như grpcurl hay là Postman gRPC hoạt động êm ái không cần thèm kéo chung nhập thêm các thẻ .proto files
    reflection.Register(srv)

    // --- Phát cờ lệnh mở trạm thu lắng nghe ---
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        log.Fatalf("mục listen: %v", err)
    }

    log.Printf("Bản doanh gRPC server đang kê tai đón khách ở ngõ :50051")

    // --- Rút chốt dọn dẹp tắt máy thật mượt mà êm dịu (Graceful shutdown) ---
    go func() {
        if err := srv.Serve(lis); err != nil {
            log.Printf("va phải lỗi serve error: %v", err)
        }
    }()

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

    log.Println("tiến hành tháo ngòi ngắt dần dàn gRPC server...")
    healthSrv.SetServingStatus("driver.v1.DriverService", grpc_health_v1.HealthCheckResponse_NOT_SERVING)
    srv.GracefulStop() // Nín thở đứng chờ cho tụi tác vụ in-flight RPCs (các chuyến gRPC đang lơ lửng dở dang chưa rớt xuống điểm đáp) hoàn tất cho sạch
    log.Println("hệ thống server đã ngừng tay")
}

// 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) {
    // Chứng chỉ bảo chứng của khách mTLS client credentials
    cert, err := tls.LoadX509KeyPair("certs/client.crt", "certs/client.key")
    if err != nil {
        return nil, nil, fmt.Errorf("cấn ở vòng 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),
        // Điều chuyển mặc định dùng round-robin gánh vác san sẻ (load balancing) đánh dạt qua dàn 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("quay số điện dial %s bị: %w", target, err)
    }

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

// Bảng biểu trình bày ví dụ cách xài 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("gặp lỗi ở khúc GetDriver error: %v", err)
        return
    }
    log.Printf("Danh tánh tay Driver %s là đang ở chế độ %s tại khoanh vùng (%f, %f)", resp.DriverId, resp.Status, resp.Latitude, resp.Longitude)
}

# Chuẩn file Dockerfile — xây tầng nhiều bước multi-stage build thu nhỏ ráo riết cục gánh dung lượng ảnh 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"]

# Lệnh file 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"

// ❌ Hỏng bét Bad: Thả rông chẳng Deadline nào cả — rủi mà cục server nó nhức đầu tự nhiên đóng băng khựng đứng im luôn (hangs), cái cục goroutine sẽ thủng đáy ngấm ngầm tứa tuôn rò rỉ rác thải đi trôi vô tận mãi mãi chả thấy ngày về
resp, err := client.GetDriver(context.Background(), req)

// ✅ Tuyệt Good: Nhớ dai kẹp chặt nhét thêm chốt cửa deadline
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
resp, err := client.GetDriver(ctx, req)

// ❌ Quá Tệ Bad: Quơ đũa gộp chùm ôm coi toàn bộ mấy thứ lỗi lầm tắp lự giống hệt một kiểu chả khác gì nhau
if err != nil {
    return fmt.Errorf("khởi báo lỗi của tụi grpc: %v", err)
}

// ✅ Sáng Láng Good: Soi chéo thật kỹ cái mã trả về status code tính bề đường lỡ nhỡ dập cầu dao thử chạy làm lại (retryability)
if err != nil {
    st, _ := status.FromError(err)
    switch st.Code() {
    case codes.NotFound:
        return nil, ErrDriverNotFound
    case codes.Unavailable, codes.ResourceExhausted:
        // Đoán được nước này sửa lại dễ Retryable — thì xịt thuốc chống sốc kéo dài backoff
        return nil, ErrRetryable
    default:
        return nil, err
    }
}

// Khuyết đi cái keepalive, bọn gác cửa ngăn lộ diện NAT firewalls êm xuôi khẽ rút nhẹ ngòi nhấn ném rớt cái bẹp (drop) các nhánh luồng gRPC rảnh rỗi quá ~4 phút lờ vờ không làm gì cả.
// Sự vụ trả lại: mấy anh bên phía client ôm cái huyễn hoặc mơ ảo thấy nghĩ rắp là mình dẫu sao vẫn đang có đính nối (connected) ngon ơ thế nhưng thực hư chả nhận lãnh được tí teo dòng dữ liệu chữ nào về tay.
// Đơn thuốc khắc chế Fix: bắt tay rèn gõ tinh chỉnh nạp cái cấu trúc keepalive đính vào lột tả ở cho cả cặp đối tác song đôi là client và phía bên anh máy chủ server (rõ ràng ở trên có show mục số Bước 4 và cái số lượng 5 nằm rành rành phía trên cao đấy).

// ❌ Trật lất Bad: Chuẩn mặc định nhà sinh đẻ (default) của tụi gRPC dính vào hàng chữ pick_first — tất tần tật cái khối luồng xe cộ mạng lưới dập ùa thẳng về chĩa mùi giáo vô nhắm đúng nguyên mỗi một em pod
conn, _ := grpc.NewClient("dns:///driver-service:50051", grpc.WithTransportCredentials(creds))

// ✅ Đúng điệu Good: cỗ bài round_robin sẽ sải cánh phân phát lố lượng công việc mâm cỗ gánh đều phân trải bao bọc lên chia ráo tẽ ra khắp giáp mặt toàn bộ đội ngũ dàn the healthy pods
conn, _ := grpc.NewClient(
    "dns:///driver-service:50051",
    grpc.WithTransportCredentials(creds),
    grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`),
)