Location Tracker (Version 0.1.0)

Author: Mohammad Hamdan

go.mod

@@ -4,6 +4,7 @@ go 1.24.5
 
 require (
 	github.com/golang/protobuf v1.5.4
+	github.com/google/uuid v1.6.0
 	github.com/gorilla/websocket v1.5.3
 	github.com/redis/go-redis/v9 v9.11.0
 	github.com/stretchr/testify v1.10.0

main.go

@@ -0,0 +1,110 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"net/http"
+	"os"
+	"os/signal"
+	"sync"
+	"syscall"
+	"time"
+
+	"github.com/MohammadAsDjangoDev/geo_tracker/src/config"
+	"github.com/MohammadAsDjangoDev/geo_tracker/src/infrastructure/repositories/redisrepo"
+	"github.com/MohammadAsDjangoDev/geo_tracker/src/infrastructure/transport/grpcserver"
+	"github.com/MohammadAsDjangoDev/geo_tracker/src/infrastructure/transport/websocket"
+	"github.com/MohammadAsDjangoDev/geo_tracker/src/usecases"
+	"github.com/redis/go-redis/v9"
+)
+
+func main() {
+	// Load config.yaml
+	cfg, err := config.LoadConfig("config.yaml")
+	if err != nil {
+		log.Fatalf("Failed to load config.yaml: %v", err)
+	}
+
+	redisAddr := fmt.Sprintf("%s:%d", cfg.Redis.Host, cfg.Redis.Port)
+	log.Printf("Connecting to Redis at %s ...", redisAddr)
+
+	redisClient := redis.NewClient(&redis.Options{
+		Addr:     redisAddr,
+		Username: cfg.Redis.Username,
+		Password: cfg.Redis.Password,
+		DB:       cfg.Redis.DB,
+	})
+
+	// Ping Redis to verify connection before proceeding
+	if err := redisClient.Ping(context.Background()).Err(); err != nil {
+		log.Fatalf("Failed to connect to Redis: %v", err)
+	}
+	log.Println("Successfully connected to Redis")
+
+	defer func() {
+		log.Println("Closing Redis connection...")
+		if err := redisClient.Close(); err != nil {
+			log.Printf("Error closing Redis connection: %v", err)
+		} else {
+			log.Println("Redis connection closed")
+		}
+	}()
+
+	// Repository and Service
+	repo := redisrepo.NewRedisTrackedItemRepository(redisClient)
+	trackingService := usecases.NewTrackingService(repo)
+
+	// WebSocket server
+	wsServer := websocket.NewServer(trackingService)
+
+	// gRPC server
+	grpcSrv, err := grpcserver.RunGRPCServer(trackingService, 50051)
+	if err != nil {
+		log.Fatalf("failed to start gRPC server: %v", err)
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	// Run WebSocket server
+	go func() {
+		defer wg.Done()
+		if err := wsServer.Start(":8080"); err != nil && err != http.ErrServerClosed {
+			log.Printf("WebSocket server error: %v", err)
+		}
+	}()
+
+	// Run gRPC server goroutine, replaced blocking select{} with channel wait
+	done := make(chan struct{})
+	go func() {
+		defer wg.Done()
+		<-done
+	}()
+
+	// Listen for termination signals (Ctrl+C)
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
+
+	<-sigChan // wait for signal
+	log.Println("Shutdown signal received, stopping servers...")
+
+	// Graceful shutdown context with timeout
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	// Shutdown WebSocket server
+	if err := wsServer.Stop(ctx); err != nil {
+		log.Printf("Error shutting down WebSocket server: %v", err)
+	}
+
+	// Shutdown gRPC server
+	grpcSrv.Stop()
+
+	// Signal gRPC goroutine to exit
+	close(done)
+
+	// Wait for goroutines to finish
+	wg.Wait()
+	log.Println("Servers stopped, exiting.")
+}

src/infrastructure/repositories/redis_tracked_item_repository.go renamed to src/infrastructure/repositories/redisrepo/redis_tracked_item_repository.go

@@ -1,4 +1,4 @@
-package repositories
+package redisrepo
 
 import (
 	"context"
@@ -40,13 +40,31 @@ func (r *RedisTrackedItemRepository) Save(ctx context.Context, item entities.Tra
 	return r.client.Set(ctx, key, data, 0).Err()
 }
 
+func (r *RedisTrackedItemRepository) Update(ctx context.Context, item entities.TrackedItem) error {
+	key := r.key(item.ID)
+	exists, err := r.client.Exists(ctx, key).Result()
+	if err != nil {
+		return err
+	}
+	if exists == 0 {
+		return irepo.ErrItemNotFound
+	}
+
+	data, err := json.Marshal(item)
+	if err != nil {
+		return err
+	}
+
+	return r.client.Set(ctx, key, data, 0).Err()
+}
+
 func (r *RedisTrackedItemRepository) UpdatePosition(ctx context.Context, id entities.ItemID, newPos entities.GeoPos) error {
 	item, err := r.FindByID(ctx, id)
 	if err != nil {
 		return err
 	}
 	item.Position = newPos
-	return r.Save(ctx, item)
+	return r.Update(ctx, item)
 }
 
 func (r *RedisTrackedItemRepository) Delete(ctx context.Context, id entities.ItemID) error {

src/infrastructure/repositories/tests/.gitignore renamed to src/infrastructure/repositories/redisrepo/tests/.gitignore

@@ -4,9 +4,6 @@ import (
 	"context"
 	"fmt"
 	"net"
-	"os"
-	"os/signal"
-	"syscall"
 
 	"github.com/MohammadAsDjangoDev/geo_tracker/src/entities"
 	"github.com/MohammadAsDjangoDev/geo_tracker/src/interfaces/grpcapi"
@@ -18,47 +15,62 @@ import (
 	"google.golang.org/protobuf/types/known/emptypb"
 )
 
+// Server struct holds grpc.Server and your existing Server implementation
 type Server struct {
 	grpcapi.UnimplementedTrackingServiceServer
 	service usecases.TrackingService
+	grpcSrv *grpc.Server
+	lis     net.Listener
 }
 
 func NewServer(service usecases.TrackingService) *Server {
-	return &Server{service: service}
+	return &Server{
+		service: service,
+	}
 }
 
-// RunGRPCServer starts the gRPC server on the specified port with the given TrackingService implementation.
-func RunGRPCServer(service usecases.TrackingService, port int) error {
+// RunGRPCServer starts the gRPC server and returns a Server instance for graceful shutdown.
+// Important: this function no longer listens for OS signals — caller (main) should handle signals
+// and call the returned Server.Stop() to perform a graceful shutdown.
+func RunGRPCServer(service usecases.TrackingService, port int) (*Server, error) {
 	lis, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
 	if err != nil {
-		return fmt.Errorf("failed to listen: %w", err)
+		return nil, fmt.Errorf("failed to listen: %w", err)
 	}
 
-	grpcServer := grpc.NewServer()
+	grpcSrv := grpc.NewServer()
 	srv := NewServer(service)
+	srv.grpcSrv = grpcSrv
+	srv.lis = lis
 
-	// Register your server implementation with the gRPC server
-	grpcapi.RegisterTrackingServiceServer(grpcServer, srv)
-
-	// Optional: enable reflection for easier debugging and tooling support (e.g., grpcurl)
-	reflection.Register(grpcServer)
+	grpcapi.RegisterTrackingServiceServer(grpcSrv, srv)
+	reflection.Register(grpcSrv)
 
-	// Graceful shutdown on SIGINT or SIGTERM
+	// Serve in a goroutine so RunGRPCServer returns immediately and caller controls lifecycle.
 	go func() {
-		c := make(chan os.Signal, 1)
-		signal.Notify(c, os.Interrupt, syscall.SIGTERM)
-		<-c
-		fmt.Println("Shutting down gRPC server...")
-		grpcServer.GracefulStop()
+		if err := grpcSrv.Serve(lis); err != nil {
+			// Serve returns a non-nil error only on failure; log it for visibility.
+			// This goroutine ends when Serve returns (e.g., after GracefulStop()).
+			fmt.Printf("gRPC server stopped serving: %v\n", err)
+		}
 	}()
 
 	fmt.Printf("gRPC server listening on port %d\n", port)
-	if err := grpcServer.Serve(lis); err != nil {
-		return fmt.Errorf("failed to serve gRPC server: %w", err)
+	return srv, nil
+}
+
+// Stop gracefully stops the gRPC server and closes the listener
+func (s *Server) Stop() {
+	if s.grpcSrv != nil {
+		s.grpcSrv.GracefulStop()
+	}
+	if s.lis != nil {
+		_ = s.lis.Close()
 	}
-	return nil
 }
 
+// RPC handlers (unchanged)
+
 func (s *Server) Register(ctx context.Context, req *grpcapi.RegisterRequest) (*emptypb.Empty, error) {
 	if req.Item == nil {
 		return nil, status.Error(codes.InvalidArgument, "item is required")

src/infrastructure/repositories/tests/redis_repository_integration_test.go renamed to src/infrastructure/repositories/redisrepo/tests/redis_repository_integration_test.go

@@ -2,8 +2,10 @@ package websocket
 
 import (
 	"context"
+	"io"
 	"log"
 	"net/http"
+	"strings"
 	"sync"
 
 	"github.com/MohammadAsDjangoDev/geo_tracker/src/entities"
@@ -16,6 +18,8 @@ type Server struct {
 	Upgrader websocket.Upgrader
 	Clients  map[entities.ItemID]*websocket.Conn
 	Mu       sync.Mutex
+
+	httpServer *http.Server
 }
 
 func NewServer(service usecases.TrackingService) *Server {
@@ -63,16 +67,26 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
 			Latitude  float64 `json:"latitude"`
 			Longitude float64 `json:"longitude"`
 		}
-		if err := conn.ReadJSON(&update); err != nil {
-			log.Println("Connection closed or error:", err)
+		err := conn.ReadJSON(&update)
+
+		if err != nil {
+			if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) {
+				log.Printf("WebSocket closed normally: %s", itemID)
+			} else if err == io.EOF {
+				log.Printf("WebSocket closed (EOF): %s", itemID)
+			} else if strings.Contains(err.Error(), "close 1006") || strings.Contains(err.Error(), "unexpected EOF") {
+				log.Printf("WebSocket closed with 1006 or unexpected EOF (abnormal but expected): %s", itemID)
+			} else {
+				log.Printf("WebSocket read error: %v", err)
+			}
 			break
 		}
 
 		pos := entities.GeoPos{
 			Latitude:  update.Latitude,
 			Longitude: update.Longitude,
 		}
-		err := s.Tracking.UpdatePosition(ctx, itemID, pos)
+		err = s.Tracking.UpdatePosition(ctx, itemID, pos)
 		if err != nil {
 			log.Println("Failed to update position:", err)
 			conn.WriteMessage(websocket.TextMessage, []byte("error: failed to update position"))
@@ -86,7 +100,33 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
 }
 
 func (s *Server) Start(addr string) error {
-	http.HandleFunc("/ws", s.HandleWebSocket)
+	mux := http.NewServeMux()
+	mux.HandleFunc("/ws", s.HandleWebSocket)
+
+	s.httpServer = &http.Server{
+		Addr:    addr,
+		Handler: mux,
+	}
+
 	log.Println("WebSocket server started on", addr)
-	return http.ListenAndServe(addr, nil)
+	return s.httpServer.ListenAndServe()
+}
+
+// Stop gracefully shuts down the HTTP server and closes all open websocket connections
+func (s *Server) Stop(ctx context.Context) error {
+	if s.httpServer != nil {
+		if err := s.httpServer.Shutdown(ctx); err != nil {
+			return err
+		}
+	}
+
+	s.Mu.Lock()
+	for id, conn := range s.Clients {
+		_ = conn.Close()
+		delete(s.Clients, id)
+	}
+	s.Mu.Unlock()
+
+	log.Println("WebSocket server stopped")
+	return nil
 }