server/handlers/event.go

@@ -16,7 +16,7 @@ import (
 )
 
 func HandleEvent(ws *websocket.Conn, message []interface{}) {
-
+	config.LimitedGoRoutine(func() {
 		if len(message) != 2 {
 			fmt.Println("Invalid EVENT message format")
 			response.SendNotice(ws, "", "Invalid EVENT message format")
@@ -64,7 +64,7 @@ func HandleEvent(ws *websocket.Conn, message []interface{}) {
 		db.StoreMongoEvent(context.TODO(), evt, ws)
 
 		fmt.Println("Event processed:", evt.ID)
-
+	})
 }
 
 func handleBlacklistAndWhitelist(ws *websocket.Conn, evt nostr.Event) bool {

server/handlers/req.go

@@ -16,41 +16,8 @@ import (
 var subscriptions = make(map[string]relay.Subscription)
 var mu sync.Mutex // Protect concurrent access to subscriptions map
 
-// RequestQueue holds incoming requests
-var RequestQueue = make(chan Request, 1000) // Adjust buffer size as needed
-
-type Request struct {
-	Ws      *websocket.Conn
-	Message []interface{}
-}
-
-// StartWorkerPool initializes a pool of worker goroutines
-func StartWorkerPool(numWorkers int) {
-	for i := 0; i < numWorkers; i++ {
-		go worker()
-	}
-}
-
-// worker processes requests from the RequestQueue
-func worker() {
-	for req := range RequestQueue {
-		processRequest(req.Ws, req.Message)
-	}
-}
-
-// HandleReq now just adds the request to the queue
 func HandleReq(ws *websocket.Conn, message []interface{}, subscriptions map[string][]relay.Filter) {
-	select {
-	case RequestQueue <- Request{Ws: ws, Message: message}:
-		// Request added to queue
-	default:
-		// Queue is full, log the dropped request
-		fmt.Println("Warning: Request queue is full, dropping request")
-	}
-}
-
-// processRequest handles the actual processing of each request
-func processRequest(ws *websocket.Conn, message []interface{}) {
+	config.LimitedGoRoutine(func() {
 		if len(message) < 3 {
 			fmt.Println("Invalid REQ message format")
 			response.SendClosed(ws, "", "invalid: invalid REQ message format")
@@ -102,7 +69,7 @@ func processRequest(ws *websocket.Conn, message []interface{}) {
 		}
 
 		// Add the new subscription or update the existing one
-	subscriptions[subID] = relay.Subscription{Filters: filters}
+		subscriptions[subID] = filters
 		fmt.Printf("Subscription updated: %s with %d filters\n", subID, len(filters))
 
 		// Query the database with filters and send back the results
@@ -142,9 +109,5 @@ func processRequest(ws *websocket.Conn, message []interface{}) {
 		}
 
 		fmt.Println("Subscription handling completed, keeping connection open.")
-}
-
-// Initialize the worker pool when your server starts
-func init() {
-	StartWorkerPool(10) // Adjust the number of workers as needed
+	})
 }