Persistance and Other Enhancements

README.md

@@ -16,6 +16,7 @@ This is a distributed, in-memory key-value store with write-ahead logging (WAL)
 - [ ] **Persistent Disk Storage**: Working on storing data to disk efficiently.
 - [ ] **Additional Endpoints**: Implementing `GET`, `UPDATE`, `DELETE`, and other operations.
 - [ ] **GraphQL & Live Queries**: Exploring GraphQL or similar solutions for reactivity.
+- [ ] Implement red black tree by myself
 
 ## Future Plans
 - **Replication & Failover**: Implement strategies for high availability.

cmd/default-config.json

@@ -19,6 +19,6 @@
 	 "inMemoryStorageThreshold": 2000,
 	 "metaDataConfig": {
 	 	 "state": 1,
-	 	 "walPath": "/var/lib/db/wal"
+	 	 "walPath": "../runtime-files/wal-storage"
 	 }
 }

cmd/init.go

@@ -1,62 +1,110 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"log"
 	"net/http"
 	"os"
+	"slices"
+	"time"
 
 	"github.com/SuperALKALINEdroiD/timelyDB/config"
 	"github.com/SuperALKALINEdroiD/timelyDB/core"
 	"github.com/SuperALKALINEdroiD/timelyDB/handlers"
-	"github.com/go-chi/chi/middleware"
-	"github.com/go-chi/chi/v5"
+	"github.com/SuperALKALINEdroiD/timelyDB/utils/common"
+	"github.com/google/uuid"
 )
 
 func initEnvironment() (*config.DatabaseConfig, error) {
-	var configPath = os.Getenv("CONFIG_PATH")
-
-	fmt.Println(configPath)
+	var configPath = os.Getenv("LOG_BASE_SETTINGS")
 
 	cfg, err := config.LoadConfig(configPath)
 	if err != nil {
 		log.Printf("Error loading configuration: %v", err)
 		return nil, err
 	}
+
 	return cfg, nil
 }
 
-func initRouter(app *core.App) *chi.Mux {
-	router := chi.NewRouter()
-	addMiddlewares(router)
-	initRoutes(router, app)
-	return router
+func GetAppPath() string {
+	return common.GetAppPath()
+}
+
+func initRouter(app *core.App) *http.ServeMux {
+	mux := http.NewServeMux()
+	initRoutes(mux, app)
+	return mux
 }
 
-func addMiddlewares(router *chi.Mux) {
-	router.Use(middleware.RealIP)
-	router.Use(middleware.RequestID)
-	router.Use(middleware.Logger)
+func initRoutes(mux *http.ServeMux, app *core.App) {
+	mux.HandleFunc("GET /data-in/health", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		fmt.Fprintln(w, "Server is running")
+	})
+
+	mux.HandleFunc("POST /data-in/upsert", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		fmt.Fprintf(w, "Upsert Endpoint WIP - Config: %+v", app)
+	})
+
+	mux.HandleFunc("POST /data-in/insert", handlers.InsertHandler(app))
+
+	mux.HandleFunc("GET /data-in/", handlers.GetValue(app))
+
+	mux.HandleFunc("POST /data-in/update", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		fmt.Fprintf(w, "Update Endpoint WIP - Config: %+v", app)
+	})
 }
 
-func initRoutes(router *chi.Mux, app *core.App) {
-	// init routes based on config ??
-	router.Route("/data-in", func(r chi.Router) {
-		r.Get("/health", func(w http.ResponseWriter, r *http.Request) {
-			w.WriteHeader(http.StatusOK)
-			fmt.Fprintln(w, "Server is running")
-		})
+func middleware(h http.Handler, m ...func(http.Handler) http.Handler) http.Handler {
+	for _, value := range slices.Backward(m) {
+		h = value(h)
+	}
 
-		r.Post("/upsert", func(w http.ResponseWriter, r *http.Request) {
-			w.WriteHeader(http.StatusOK)
-			fmt.Fprintf(w, "Upsert Endpoint WIP - Config: %+v", app)
-		})
+	return h
+}
 
-		r.Post("/insert", handlers.InsertHandler(app))
+func realIP(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if ip := r.Header.Get("X-Real-IP"); ip != "" {
+			r.RemoteAddr = ip
+		} else if ip := r.Header.Get("X-Forwarded-For"); ip != "" {
+			r.RemoteAddr = ip
+		}
+		next.ServeHTTP(w, r)
+	})
+}
 
-		r.Post("/update", func(w http.ResponseWriter, r *http.Request) {
-			w.WriteHeader(http.StatusOK)
-			fmt.Fprintf(w, "Update Endpoint WIP - Config: %+v", app)
-		})
+func requestID(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		id := r.Header.Get("X-Request-ID")
+		if id == "" {
+			id = uuid.NewString()
+		}
+		w.Header().Set("X-Request-ID", id)
+		ctx := context.WithValue(r.Context(), "requestID", id)
+		next.ServeHTTP(w, r.WithContext(ctx))
 	})
 }
+
+func requestLogger(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		start := time.Now()
+		rw := &statusResponseWriter{ResponseWriter: w, status: http.StatusOK}
+		next.ServeHTTP(rw, r)
+		log.Printf("%s %s %d %s", r.Method, r.URL.Path, rw.status, time.Since(start))
+	})
+}
+
+type statusResponseWriter struct {
+	http.ResponseWriter
+	status int
+}
+
+func (rw *statusResponseWriter) WriteHeader(code int) {
+	rw.status = code
+	rw.ResponseWriter.WriteHeader(code)
+}

cmd/main.go

@@ -7,14 +7,17 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
-	"runtime/pprof"
+	"path/filepath"
 	"syscall"
 	"time"
 
 	"github.com/SuperALKALINEdroiD/timelyDB/core"
+	"github.com/SuperALKALINEdroiD/timelyDB/utils/common"
 	"github.com/SuperALKALINEdroiD/timelyDB/utils/logs"
 	"github.com/SuperALKALINEdroiD/timelyDB/utils/nodes"
 	"github.com/SuperALKALINEdroiD/timelyDB/utils/storage"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 )
 
 func main() {
@@ -24,10 +27,6 @@ func main() {
 	}
 	defer f.Close()
 
-	if err := pprof.StartCPUProfile(f); err != nil {
-		panic(err)
-	}
-	defer pprof.StopCPUProfile()
 	ctx, cancel := context.WithCancel(context.Background())
 
 	signalChannel := make(chan os.Signal, 1)
@@ -45,24 +44,54 @@ func main() {
 		panic("error while loading config")
 	}
 
-	grpcNodes, nodeHashInfo := nodes.LoadServers(ctx, config)
 	wal := &storage.LocalWAL{}
-	wal.Connect("wal-storage")
+	appPath := common.GetAppPath()
+	wal.Connect(filepath.Join(appPath, config.MetaDataConfig.WALName))
+
+	grpcNodes, nodeHashInfo := nodes.LoadServers(ctx, config, wal)
+
+	storageNodesIndex := make(map[string]*nodes.Node, len(grpcNodes))
+	nodeClients := make(map[string]nodes.NodeServiceClient, len(grpcNodes))
+	nodeConns := make(map[string]*grpc.ClientConn, len(grpcNodes))
+
+	for _, n := range grpcNodes {
+		if n == nil {
+			continue
+		}
+		storageNodesIndex[n.ID] = n
+		conn, connErr := grpc.NewClient(n.Address, grpc.WithTransportCredentials(insecure.NewCredentials()))
+		if connErr != nil {
+			log.Fatalf("failed to create gRPC client for node %s: %v", n.ID, connErr)
+		}
+		nodeClients[n.ID] = nodes.NewNodeServiceClient(conn)
+		nodeConns[n.ID] = conn
+	}
 
 	app := &core.App{
 		Config:       config,
 		Nodes:        grpcNodes,
+		NodeByID:     storageNodesIndex,
+		NodeClients:  nodeClients,
+		NodeConns:    nodeConns,
 		NodeHashInfo: nodeHashInfo,
 		WAL:          wal,
 	}
 
 	logs.ReplayLogs(app)
 
 	router := initRouter(app)
+	handler := middleware(router, realIP, requestID, requestLogger)
 
 	serverAddress := fmt.Sprintf(":%d", app.Config.Port)
-	log.Printf("Starting server on %s", serverAddress)
-	server := &http.Server{Addr: ":7001", Handler: router}
+	log.Printf("Starting %s server on %s", config.StoreName, serverAddress)
+	server := &http.Server{
+		Addr:              serverAddress,
+		Handler:           handler,
+		ReadHeaderTimeout: 5 * time.Second,
+		ReadTimeout:       15 * time.Second,
+		WriteTimeout:      15 * time.Second,
+		IdleTimeout:       60 * time.Second,
+	}
 
 	go func() {
 		log.Printf("Starting to listen on %s", serverAddress)
@@ -74,13 +103,23 @@ func main() {
 	<-ctx.Done()
 	log.Println("Shutting down main server...")
 
+	if err := app.WAL.Flush(); err != nil {
+		log.Printf("WAL flush on shutdown failed: %v", err)
+	}
+
 	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer shutdownCancel()
 
 	if err := server.Shutdown(shutdownCtx); err != nil {
 		log.Fatalf("Server shutdown failed: %v", err)
 	}
 
+	for nodeID, conn := range app.NodeConns {
+		if err := conn.Close(); err != nil {
+			log.Printf("failed to close gRPC client for node %s: %v", nodeID, err)
+		}
+	}
+
 	log.Println("Exiting, Bye!")
 
 }