1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238

package main

import (
    "github.com/gin-gonic/gin"
    "github.com/gin-contrib/cors"
    "github.com/gin-contrib/helmet"
    "github.com/ulule/limiter/v3"
    "github.com/ulule/limiter/v3/drivers/store/memory"
    "github.com/go-redis/redis/v8"
    "github.com/prometheus/client_golang/prometheus"
    "go.uber.org/zap"
    "net/http"
    "time"
)

type Config struct {
    Port            string
    RedisAddr       string
    RateLimit       int
    RateWindow      time.Duration
    EnableMetrics   bool
    EnableTracing   bool
}

type Server struct {
    router  *gin.Engine
    redis   *redis.Client
    limiter *limiter.Limiter
    logger  *zap.Logger
    metrics *prometheus.Registry
}

func NewServer(cfg *Config) (*Server, error) {
    // Initialize Redis
    rdb := redis.NewClient(&redis.Options{
        Addr: cfg.RedisAddr,
    })

    // Initialize rate limiter
    rate := limiter.Rate{
        Period: cfg.RateWindow,
        Limit:  int64(cfg.RateLimit),
    }
    store := memory.NewStore()
    limiter := limiter.New(store, rate)

    // Initialize logger
    logger, err := zap.NewProduction()
    if err != nil {
        return nil, err
    }

    // Initialize metrics
    metrics := prometheus.NewRegistry()

    // Create server
    s := &Server{
        router:  gin.Default(),
        redis:   rdb,
        limiter: limiter,
        logger:  logger,
        metrics: metrics,
    }

    // Setup middleware
    s.setupMiddleware()

    // Setup routes
    s.setupRoutes()

    return s, nil
}

func (s *Server) setupMiddleware() {
    // CORS
    s.router.Use(cors.New(cors.Config{
        AllowOrigins:     []string{"*"},
        AllowMethods:     []string{"GET", "POST", "PUT", "DELETE", "OPTIONS"},
        AllowHeaders:     []string{"Origin", "Content-Type", "Authorization"},
        ExposeHeaders:    []string{"Content-Length"},
        AllowCredentials: true,
        MaxAge:           12 * time.Hour,
    }))

    // Security headers
    s.router.Use(helmet.Default())

    // Rate limiting
    s.router.Use(s.rateLimitMiddleware())

    // Logging
    s.router.Use(s.loggingMiddleware())

    // Metrics
    s.router.Use(s.metricsMiddleware())

    // Recovery
    s.router.Use(gin.Recovery())
}

func (s *Server) setupRoutes() {
    // Health check
    s.router.GET("/health", s.healthCheck)

    // API routes
    api := s.router.Group("/api/v1")
    {
        // User routes
        users := api.Group("/users")
        {
            users.GET("", s.getUsers)
            users.GET("/:id", s.getUser)
            users.POST("", s.createUser)
            users.PUT("/:id", s.updateUser)
            users.DELETE("/:id", s.deleteUser)
        }

        // Product routes
        products := api.Group("/products")
        {
            products.GET("", s.getProducts)
            products.GET("/:id", s.getProduct)
            products.POST("", s.createProduct)
            products.PUT("/:id", s.updateProduct)
            products.DELETE("/:id", s.deleteProduct)
        }
    }
}

func (s *Server) rateLimitMiddleware() gin.HandlerFunc {
    return func(c *gin.Context) {
        ip := c.ClientIP()
        context, err := s.limiter.Get(c, ip)
        if err != nil {
            c.JSON(http.StatusInternalServerError, gin.H{"error": "rate limit error"})
            c.Abort()
            return
        }

        if context.Reached {
            c.JSON(http.StatusTooManyRequests, gin.H{
                "error": "rate limit exceeded",
                "retry_after": context.Reset,
            })
            c.Abort()
            return
        }

        c.Next()
    }
}

func (s *Server) loggingMiddleware() gin.HandlerFunc {
    return func(c *gin.Context) {
        start := time.Now()
        path := c.Request.URL.Path
        query := c.Request.URL.RawQuery

        c.Next()

        latency := time.Since(start)
        status := c.Writer.Status()
        clientIP := c.ClientIP()
        method := c.Request.Method
        userAgent := c.Request.UserAgent()

        s.logger.Info("request completed",
            zap.String("path", path),
            zap.String("query", query),
            zap.Int("status", status),
            zap.Duration("latency", latency),
            zap.String("ip", clientIP),
            zap.String("method", method),
            zap.String("user-agent", userAgent),
        )
    }
}

func (s *Server) metricsMiddleware() gin.HandlerFunc {
    httpRequests := prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "http_requests_total",
            Help: "Total number of HTTP requests",
        },
        []string{"method", "path", "status"},
    )

    httpDuration := prometheus.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "http_request_duration_seconds",
            Help:    "HTTP request duration in seconds",
            Buckets: prometheus.DefBuckets,
        },
        []string{"method", "path"},
    )

    s.metrics.MustRegister(httpRequests, httpDuration)

    return func(c *gin.Context) {
        start := time.Now()
        path := c.Request.URL.Path

        c.Next()

        status := fmt.Sprintf("%d", c.Writer.Status())
        duration := time.Since(start).Seconds()

        httpRequests.WithLabelValues(c.Request.Method, path, status).Inc()
        httpDuration.WithLabelValues(c.Request.Method, path).Observe(duration)
    }
}

func (s *Server) healthCheck(c *gin.Context) {
    c.JSON(http.StatusOK, gin.H{
        "status": "healthy",
        "time":   time.Now(),
    })
}

func main() {
    cfg := &Config{
        Port:            ":8080",
        RedisAddr:       "localhost:6379",
        RateLimit:       100,
        RateWindow:      15 * time.Minute,
        EnableMetrics:   true,
        EnableTracing:   true,
    }

    server, err := NewServer(cfg)
    if err != nil {
        log.Fatal(err)
    }

    if err := server.router.Run(cfg.Port); err != nil {
        log.Fatal(err)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

syntax = "proto3";

package user;

import "google/protobuf/timestamp.proto";
import "google/protobuf/empty.proto";
import "google/api/annotations.proto";

service UserService {
    rpc GetUser(GetUserRequest) returns (User) {
        option (google.api.http) = {
            get: "/v1/users/{id}"
        };
    }
    
    rpc CreateUser(CreateUserRequest) returns (User) {
        option (google.api.http) = {
            post: "/v1/users"
            body: "*"
        };
    }
    
    rpc UpdateUser(UpdateUserRequest) returns (User) {
        option (google.api.http) = {
            put: "/v1/users/{id}"
            body: "*"
        };
    }
    
    rpc DeleteUser(DeleteUserRequest) returns (google.protobuf.Empty) {
        option (google.api.http) = {
            delete: "/v1/users/{id}"
        };
    }
    
    rpc ListUsers(ListUsersRequest) returns (stream User) {
        option (google.api.http) = {
            get: "/v1/users"
        };
    }
}

message User {
    string id = 1;
    string name = 2;
    string email = 3;
    google.protobuf.Timestamp created_at = 4;
    google.protobuf.Timestamp updated_at = 5;
    repeated string roles = 6;
    map<string, string> metadata = 7;
}

message GetUserRequest {
    string id = 1;
}

message CreateUserRequest {
    string name = 1;
    string email = 2;
    string password = 3;
    repeated string roles = 4;
    map<string, string> metadata = 5;
}

message UpdateUserRequest {
    string id = 1;
    string name = 2;
    string email = 3;
    repeated string roles = 4;
    map<string, string> metadata = 5;
}

message DeleteUserRequest {
    string id = 1;
}

message ListUsersRequest {
    int32 page_size = 1;
    string page_token = 2;
    string filter = 3;
    string order_by = 4;
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248

package main

import (
    "context"
    "log"
    "net"
    "time"

    pb "path/to/generated/proto"
    "github.com/go-redis/redis/v8"
    "github.com/prometheus/client_golang/prometheus"
    "go.uber.org/zap"
    grpcmiddleware "github.com/grpc-ecosystem/go-grpc-middleware/v2"
    grpclogging "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/logging"
    grpcrecovery "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/recovery"
    zapprovider "github.com/grpc-ecosystem/go-grpc-middleware/providers/zap"
    "google.golang.org/grpc"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/reflection"
    "google.golang.org/grpc/status"
    "google.golang.org/protobuf/types/known/emptypb"
    "google.golang.org/protobuf/types/known/timestamppb"
)

type server struct {
    pb.UnimplementedUserServiceServer
    redis   *redis.Client
    logger  *zap.Logger
    metrics *prometheus.Registry
}

func NewServer(redisAddr string) (*server, error) {
    // Initialize Redis
    rdb := redis.NewClient(&redis.Options{
        Addr: redisAddr,
    })

    // Initialize logger
    logger, err := zap.NewProduction()
    if err != nil {
        return nil, err
    }

    // Initialize metrics
    metrics := prometheus.NewRegistry()

    return &server{
        redis:   rdb,
        logger:  logger,
        metrics: metrics,
    }, nil
}

func (s *server) GetUser(ctx context.Context, req *pb.GetUserRequest) (*pb.User, error) {
    // Log request
    s.logger.Info("getting user",
        zap.String("id", req.Id),
    )

    // Get user from Redis
    user, err := s.redis.HGetAll(ctx, "user:"+req.Id).Result()
    if err != nil {
        if err == redis.Nil {
            return nil, status.Error(codes.NotFound, "user not found")
        }
        return nil, status.Error(codes.Internal, err.Error())
    }

    // Convert to proto
    return &pb.User{
        Id:        user["id"],
        Name:      user["name"],
        Email:     user["email"],
        CreatedAt: parseTime(user["created_at"]),
        UpdatedAt: parseTime(user["updated_at"]),
    }, nil
}

func (s *server) CreateUser(ctx context.Context, req *pb.CreateUserRequest) (*pb.User, error) {
    // Log request
    s.logger.Info("creating user",
        zap.String("email", req.Email),
    )

    // Validate request
    if err := validateCreateUserRequest(req); err != nil {
        return nil, status.Error(codes.InvalidArgument, err.Error())
    }

    // Create user
    now := time.Now()
    user := &pb.User{
        Id:        generateID(),
        Name:      req.Name,
        Email:     req.Email,
        CreatedAt: timestamppb.New(now),
        UpdatedAt: timestamppb.New(now),
        Roles:     req.Roles,
        Metadata:  req.Metadata,
    }

    // Save to Redis
    if err := s.saveUser(ctx, user); err != nil {
        return nil, status.Error(codes.Internal, err.Error())
    }

    return user, nil
}

func (s *server) UpdateUser(ctx context.Context, req *pb.UpdateUserRequest) (*pb.User, error) {
    // Log request
    s.logger.Info("updating user",
        zap.String("id", req.Id),
    )

    // Get existing user
    existing, err := s.GetUser(ctx, &pb.GetUserRequest{Id: req.Id})
    if err != nil {
        return nil, err
    }

    // Update fields
    if req.Name != "" {
        existing.Name = req.Name
    }
    if req.Email != "" {
        existing.Email = req.Email
    }
    if req.Roles != nil {
        existing.Roles = req.Roles
    }
    if req.Metadata != nil {
        existing.Metadata = req.Metadata
    }
    existing.UpdatedAt = timestamppb.New(time.Now())

    // Save to Redis
    if err := s.saveUser(ctx, existing); err != nil {
        return nil, status.Error(codes.Internal, err.Error())
    }

    return existing, nil
}

func (s *server) DeleteUser(ctx context.Context, req *pb.DeleteUserRequest) (*emptypb.Empty, error) {
    // Log request
    s.logger.Info("deleting user",
        zap.String("id", req.Id),
    )

    // Delete from Redis
    if err := s.redis.Del(ctx, "user:"+req.Id).Err(); err != nil {
        return nil, status.Error(codes.Internal, err.Error())
    }

    return &emptypb.Empty{}, nil
}

func (s *server) ListUsers(req *pb.ListUsersRequest, stream pb.UserService_ListUsersServer) error {
    // Log request
    s.logger.Info("listing users",
        zap.Int32("page_size", req.PageSize),
        zap.String("page_token", req.PageToken),
    )

    // Get users from Redis
    keys, err := s.redis.Keys(stream.Context(), "user:*").Result()
    if err != nil {
        return status.Error(codes.Internal, err.Error())
    }

    // Stream users
    for _, key := range keys {
        user, err := s.redis.HGetAll(stream.Context(), key).Result()
        if err != nil {
            continue
        }

        if err := stream.Send(&pb.User{
            Id:        user["id"],
            Name:      user["name"],
            Email:     user["email"],
            CreatedAt: parseTime(user["created_at"]),
            UpdatedAt: parseTime(user["updated_at"]),
        }); err != nil {
            return status.Error(codes.Internal, err.Error())
        }
    }

    return nil
}

func (s *server) saveUser(ctx context.Context, user *pb.User) error {
    return s.redis.HSet(ctx, "user:"+user.Id, map[string]interface{}{
        "id":         user.Id,
        "name":       user.Name,
        "email":      user.Email,
        "created_at": user.CreatedAt.AsTime().Format(time.RFC3339),
        "updated_at": user.UpdatedAt.AsTime().Format(time.RFC3339),
    }).Err()
}

func main() {
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        log.Fatalf("failed to listen: %v", err)
    }

    s, err := NewServer("localhost:6379")
    if err != nil {
        log.Fatalf("failed to create server: %v", err)
    }

    // Minimal interceptor chain: logging + recovery
    interceptorLogger := zapprovider.InterceptorLogger(s.logger)
    grpcServer := grpc.NewServer(
        grpc.UnaryInterceptor(grpcmiddleware.ChainUnaryServer(
            grpclogging.UnaryServerInterceptor(interceptorLogger),
            grpcrecovery.UnaryServerInterceptor(),
        )),
        grpc.StreamInterceptor(grpcmiddleware.ChainStreamServer(
            grpclogging.StreamServerInterceptor(interceptorLogger),
            grpcrecovery.StreamServerInterceptor(),
        )),
    )

    pb.RegisterUserServiceServer(grpcServer, s)
    reflection.Register(grpcServer)

    if err := grpcServer.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

// --- helpers to make the example self-contained ---
func generateID() string { return time.Now().UTC().Format("20060102150405.000000000") }

func validateCreateUserRequest(req *pb.CreateUserRequest) error {
    if req.GetEmail() == "" || req.GetName() == "" { return fmt.Errorf("name and email are required") }
    return nil
}

func parseTime(s string) *timestamppb.Timestamp {
    if s == "" { return timestamppb.New(time.Time{}) }
    t, err := time.Parse(time.RFC3339, s)
    if err != nil { return timestamppb.New(time.Time{}) }
    return timestamppb.New(t)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

package main

import (
    "github.com/99designs/gqlgen/graphql"
    "github.com/99designs/gqlgen/graphql/handler"
    "github.com/99designs/gqlgen/graphql/playground"
    "github.com/gin-gonic/gin"
    "github.com/vektah/gqlparser/v2"
    "github.com/vektah/gqlparser/v2/ast"
)

// Schema definition
var schema = `
    type User {
        id: ID!
        name: String!
        email: String!
        posts: [Post!]
    }
    
    type Post {
        id: ID!
        title: String!
        content: String!
        author: User!
    }
    
    type Query {
        user(id: ID!): User
        users: [User!]!
        post(id: ID!): Post
        posts: [Post!]!
    }
    
    type Mutation {
        createUser(input: CreateUserInput!): User!
        createPost(input: CreatePostInput!): Post!
    }
    
    input CreateUserInput {
        name: String!
        email: String!
    }
    
    input CreatePostInput {
        title: String!
        content: String!
        authorId: ID!
    }
`

// Resolvers
type Resolver struct {
    db *Database
}

func (r *Resolver) Query() QueryResolver {
    return &queryResolver{r}
}

func (r *Resolver) Mutation() MutationResolver {
    return &mutationResolver{r}
}

type queryResolver struct{ *Resolver }
type mutationResolver struct{ *Resolver }

func (r *queryResolver) User(ctx context.Context, id string) (*User, error) {
    return r.db.GetUser(id)
}

func (r *queryResolver) Users(ctx context.Context) ([]*User, error) {
    return r.db.GetUsers()
}

func (r *mutationResolver) CreateUser(ctx context.Context, input CreateUserInput) (*User, error) {
    return r.db.CreateUser(input)
}

func main() {
    r := gin.Default()
    
    // GraphQL playground
    r.GET("/", playground.Handler("GraphQL playground", "/query"))
    
    // GraphQL handler
    r.POST("/query", func(c *gin.Context) {
        h := handler.NewDefaultServer(generated.NewExecutableSchema(generated.Config{
            Resolvers: &Resolver{db: NewDatabase()},
        }))
        h.ServeHTTP(c.Writer, c.Request)
    })
    
    r.Run(":4000")
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

package main

import (
    "context"
    "log"
    "net/http"
    "time"

    "github.com/99designs/gqlgen/graphql"
    "github.com/99designs/gqlgen/graphql/handler"
    "github.com/99designs/gqlgen/graphql/playground"
    "github.com/gorilla/websocket"
)

type Message struct {
    ID        string    `json:"id"`
    Content   string    `json:"content"`
    UserID    string    `json:"userId"`
    CreatedAt time.Time `json:"createdAt"`
}

type Subscription struct {
    messageAdded chan *Message
}

func (s *Subscription) MessageAdded(ctx context.Context) (<-chan *Message, error) {
    return s.messageAdded, nil
}

func main() {
    srv := handler.NewDefaultServer(generated.NewExecutableSchema(generated.Config{
        Resolvers: &Resolver{
            subscription: &Subscription{
                messageAdded: make(chan *Message),
            },
        },
    }))

    http.Handle("/", playground.Handler("GraphQL playground", "/query"))
    http.Handle("/query", srv)

    log.Fatal(http.ListenAndServe(":8080", nil))
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "google.golang.org/grpc"
    pb "your-project/proto"
)

func benchmarkGRPC() {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    conn, err := grpc.DialContext(ctx, "localhost:50051", grpc.WithTransportCredentials(insecure.NewCredentials()))
    if err != nil {
        log.Fatalf("Failed to connect: %v", err)
    }
    defer conn.Close()

    client := pb.NewYourServiceClient(conn)
    ctx = context.Background()

    start := time.Now()
    for i := 0; i < 1000; i++ {
        _, err := client.YourMethod(ctx, &pb.Request{})
        if err != nil {
            log.Printf("Error: %v", err)
        }
    }
    duration := time.Since(start)
    fmt.Printf("gRPC Benchmark avg/op: %v\n", duration/time.Duration(1000))
}

func benchmarkREST() {
    // Similar implementation for REST
}

func main() {
    benchmarkGRPC()
    benchmarkREST()
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162

package main

import (
    "context"
    "encoding/json"
    "errors"
    "log"
    "net/http"
    "time"

    "github.com/go-redis/redis/v8"
    "github.com/golang-jwt/jwt"
    "github.com/gorilla/websocket"
)

var upgrader = websocket.Upgrader{
    ReadBufferSize:  1024,
    WriteBufferSize: 1024,
    CheckOrigin: func(r *http.Request) bool {
        return true // For development
    },
}

type Client struct {
    conn     *websocket.Conn
    userID   string
    send     chan []byte
    redis    *redis.Client
}

type Message struct {
    Type      string          `json:"type"`
    UserID    string          `json:"userId"`
    Content   string          `json:"content,omitempty"`
    Title     string          `json:"title,omitempty"`
    Message   string          `json:"message,omitempty"`
    Status    string          `json:"status,omitempty"`
    Timestamp int64           `json:"timestamp"`
}

func main() {
    // Redis connection
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })

    // WebSocket endpoint
    http.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
        // Token validation
        token := r.Header.Get("Sec-WebSocket-Protocol")
        claims, err := validateToken(token)
        if err != nil {
            http.Error(w, "Unauthorized", http.StatusUnauthorized)
            return
        }

        // Upgrade to WebSocket connection
        conn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
            log.Println(err)
            return
        }

        // Create new client
        client := &Client{
            conn:   conn,
            userID: claims.UserID,
            send:   make(chan []byte, 256),
            redis:  rdb,
        }

        // Track and manage client
        clients[client] = struct{}{}
        client.redis.SAdd(context.Background(), "online_users", client.userID)
        
        // Manage client
        go client.readPump()
        go client.writePump()
    })

    log.Fatal(http.ListenAndServe(":8080", nil))
}

var clients = make(map[*Client]struct{})

func (c *Client) readPump() {
    defer func() {
        c.conn.Close()
        delete(clients, c)
        c.redis.SRem(context.Background(), "online_users", c.userID)
    }()

    for {
        _, message, err := c.conn.ReadMessage()
        if err != nil {
            if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
                log.Printf("error: %v", err)
            }
            break
        }

        // Process message
        var msg Message
        if err := json.Unmarshal(message, &msg); err != nil {
            log.Printf("error: %v", err)
            continue
        }

        // Handle message based on type
        switch msg.Type {
        case "chat":
            handleChatMessage(c, msg)
        case "notification":
            handleNotification(c, msg)
        case "status":
            handleStatusUpdate(c, msg)
        default:
            log.Printf("unknown message type: %s", msg.Type)
        }
    }
}

// validateToken is a minimal placeholder; replace with your JWT validation.
type tokenClaims struct { UserID string }

func validateToken(raw string) (*tokenClaims, error) {
    if raw == "" { return nil, errors.New("empty token") }
    // For demo purposes only; parse/verify JWT in production
    return &tokenClaims{UserID: raw}, nil
}

func (c *Client) writePump() {
    ticker := time.NewTicker(30 * time.Second)
    defer func() {
        ticker.Stop()
        c.conn.Close()
    }()

    for {
        select {
        case message, ok := <-c.send:
            if !ok {
                c.conn.WriteMessage(websocket.CloseMessage, []byte{})
                return
            }

            w, err := c.conn.NextWriter(websocket.TextMessage)
            if err != nil {
                return
            }
            w.Write(message)

            if err := w.Close(); err != nil {
                return
            }
        case <-ticker.C:
            if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
                return
            }
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

package main

import (
	"context"
	"log"
	"net"
	"net/http"
	"os"

	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/protobuf/encoding/protojson"

	"path/to/generated/proto"
)

func main() {
	lis, err := net.Listen("tcp", ":50051")
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}

	s := grpc.NewServer()
	pb.RegisterUserServiceServer(s, &server{})

	go func() {
		log.Printf("starting gRPC server on %s", lis.Addr().String())
		if err := s.Serve(lis); err != nil {
			log.Fatalf("failed to serve: %v", err)
		}
	}()

	conn, err := grpc.Dial(
		lis.Addr().String(),
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	)
	if err != nil {
		log.Fatalf("failed to dial server: %v", err)
	}
	defer conn.Close()

	gwmux := runtime.NewServeMux(
		runtime.WithMarshalerOption(runtime.MIMEWildcard, &runtime.JSONPb{
			MarshalOptions: protojson.MarshalOptions{
				UseProtoNames: true,
			},
		}),
	)

	err = pb.RegisterUserServiceHandler(context.Background(), gwmux, conn)
	if err != nil {
		log.Fatalf("failed to register gateway: %v", err)
	}

	mux := http.NewServeMux()
	mux.Handle("/", gwmux)

	log.Printf("starting HTTP server on %s", ":8080")
	if err := http.ListenAndServe(":8080", mux); err != nil {
		log.Fatalf("failed to serve: %v", err)
	}
}