server/handlers/event.go

@@ -16,55 +16,55 @@ 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")
+			return
+		}
 
-	if len(message) != 2 {
+		eventData, ok := message[1].(map[string]interface{})
-		fmt.Println("Invalid EVENT message format")
+		if !ok {
-		response.SendNotice(ws, "", "Invalid EVENT message format")
+			fmt.Println("Invalid event data format")
-		return
+			response.SendNotice(ws, "", "Invalid event data format")
-	}
+			return
+		}
+		eventBytes, err := json.Marshal(eventData)
+		if err != nil {
+			fmt.Println("Error marshaling event data:", err)
+			response.SendNotice(ws, "", "Error marshaling event data")
+			return
+		}
 
-	eventData, ok := message[1].(map[string]interface{})
+		var evt nostr.Event
-	if !ok {
+		err = json.Unmarshal(eventBytes, &evt)
-		fmt.Println("Invalid event data format")
+		if err != nil {
-		response.SendNotice(ws, "", "Invalid event data format")
+			fmt.Println("Error unmarshaling event data:", err)
-		return
+			response.SendNotice(ws, "", "Error unmarshaling event data")
-	}
+			return
-	eventBytes, err := json.Marshal(eventData)
+		}
-	if err != nil {
-		fmt.Println("Error marshaling event data:", err)
-		response.SendNotice(ws, "", "Error marshaling event data")
-		return
-	}
 
-	var evt nostr.Event
+		// Signature check moved here
-	err = json.Unmarshal(eventBytes, &evt)
+		if !utils.CheckSignature(evt) {
-	if err != nil {
+			response.SendOK(ws, evt.ID, false, "invalid: signature verification failed")
-		fmt.Println("Error unmarshaling event data:", err)
+			return
-		response.SendNotice(ws, "", "Error unmarshaling event data")
+		}
-		return
-	}
 
-	// Signature check moved here
+		eventSize := len(eventBytes) // Calculate event size
-	if !utils.CheckSignature(evt) {
-		response.SendOK(ws, evt.ID, false, "invalid: signature verification failed")
-		return
-	}
 
-	eventSize := len(eventBytes) // Calculate event size
+		if !handleBlacklistAndWhitelist(ws, evt) {
+			return
+		}
 
-	if !handleBlacklistAndWhitelist(ws, evt) {
+		if !handleRateAndSizeLimits(ws, evt, eventSize) {
-		return
+			return
-	}
+		}
 
-	if !handleRateAndSizeLimits(ws, evt, eventSize) {
+		// This is where I'll handle storage for multiple database types in the future
-		return
+		db.StoreMongoEvent(context.TODO(), evt, ws)
-	}
-
-	// This is where I'll handle storage for multiple database types in the future
-	db.StoreMongoEvent(context.TODO(), evt, ws)
-
-	fmt.Println("Event processed:", evt.ID)
 
+		fmt.Println("Event processed:", evt.ID)
+	})
 }
 
 func handleBlacklistAndWhitelist(ws *websocket.Conn, evt nostr.Event) bool {

server/handlers/req.go

@@ -16,135 +16,98 @@ 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 {
+	config.LimitedGoRoutine(func() {
-	case RequestQueue <- Request{Ws: ws, Message: message}:
+		if len(message) < 3 {
-		// Request added to queue
+			fmt.Println("Invalid REQ message format")
-	default:
+			response.SendClosed(ws, "", "invalid: invalid REQ message format")
-		// Queue is full, log the dropped request
+			return
-		fmt.Println("Warning: Request queue is full, dropping request")
-	}
-}
-
-// processRequest handles the actual processing of each request
-func processRequest(ws *websocket.Conn, message []interface{}) {
-	if len(message) < 3 {
-		fmt.Println("Invalid REQ message format")
-		response.SendClosed(ws, "", "invalid: invalid REQ message format")
-		return
-	}
-
-	subID, ok := message[1].(string)
-	if !ok {
-		fmt.Println("Invalid subscription ID format")
-		response.SendClosed(ws, "", "invalid: invalid subscription ID format")
-		return
-	}
-
-	mu.Lock()
-	defer mu.Unlock()
-
-	// Check the current number of subscriptions for the client
-	if len(subscriptions) >= config.GetConfig().Server.MaxSubscriptionsPerClient {
-		// Find and remove the oldest subscription (FIFO)
-		var oldestSubID string
-		for id := range subscriptions {
-			oldestSubID = id
-			break
 		}
-		delete(subscriptions, oldestSubID)
-		fmt.Println("Dropped oldest subscription:", oldestSubID)
-	}
 
-	// Prepare filters based on the incoming message
+		subID, ok := message[1].(string)
-	filters := make([]relay.Filter, len(message)-2)
-	for i, filter := range message[2:] {
-		filterData, ok := filter.(map[string]interface{})
 		if !ok {
-			fmt.Println("Invalid filter format")
+			fmt.Println("Invalid subscription ID format")
-			response.SendClosed(ws, subID, "invalid: invalid filter format")
+			response.SendClosed(ws, "", "invalid: invalid subscription ID format")
 			return
 		}
 
-		var f relay.Filter
+		mu.Lock()
-		f.IDs = utils.ToStringArray(filterData["ids"])
+		defer mu.Unlock()
-		f.Authors = utils.ToStringArray(filterData["authors"])
-		f.Kinds = utils.ToIntArray(filterData["kinds"])
-		f.Tags = utils.ToTagsMap(filterData["tags"])
-		f.Since = utils.ToTime(filterData["since"])
-		f.Until = utils.ToTime(filterData["until"])
-		f.Limit = utils.ToInt(filterData["limit"])
 
-		filters[i] = f
+		// Check the current number of subscriptions for the client
-	}
+		if len(subscriptions) >= config.GetConfig().Server.MaxSubscriptionsPerClient {
+			// Find and remove the oldest subscription (FIFO)
+			var oldestSubID string
+			for id := range subscriptions {
+				oldestSubID = id
+				break
+			}
+			delete(subscriptions, oldestSubID)
+			fmt.Println("Dropped oldest subscription:", oldestSubID)
+		}
 
-	// Add the new subscription or update the existing one
+		// Prepare filters based on the incoming message
-	subscriptions[subID] = relay.Subscription{Filters: filters}
+		filters := make([]relay.Filter, len(message)-2)
-	fmt.Printf("Subscription updated: %s with %d filters\n", subID, len(filters))
+		for i, filter := range message[2:] {
+			filterData, ok := filter.(map[string]interface{})
+			if !ok {
+				fmt.Println("Invalid filter format")
+				response.SendClosed(ws, subID, "invalid: invalid filter format")
+				return
+			}
 
-	// Query the database with filters and send back the results
+			var f relay.Filter
-	queriedEvents, err := db.QueryEvents(filters, db.GetClient(), "grain")
+			f.IDs = utils.ToStringArray(filterData["ids"])
-	if err != nil {
+			f.Authors = utils.ToStringArray(filterData["authors"])
-		fmt.Println("Error querying events:", err)
+			f.Kinds = utils.ToIntArray(filterData["kinds"])
-		response.SendClosed(ws, subID, "error: could not query events")
+			f.Tags = utils.ToTagsMap(filterData["tags"])
-		return
+			f.Since = utils.ToTime(filterData["since"])
-	}
+			f.Until = utils.ToTime(filterData["until"])
+			f.Limit = utils.ToInt(filterData["limit"])
 
-	// Log the number of events retrieved
+			filters[i] = f
-	fmt.Printf("Retrieved %d events for subscription %s\n", len(queriedEvents), subID)
+		}
-	if len(queriedEvents) == 0 {
-		fmt.Printf("No matching events found for subscription %s\n", subID)
-	}
 
-	// Send each event back to the client
+		// Add the new subscription or update the existing one
-	for _, evt := range queriedEvents {
+		subscriptions[subID] = filters
-		msg := []interface{}{"EVENT", subID, evt}
+		fmt.Printf("Subscription updated: %s with %d filters\n", subID, len(filters))
-		msgBytes, _ := json.Marshal(msg)
+
-		err = websocket.Message.Send(ws, string(msgBytes))
+		// Query the database with filters and send back the results
+		queriedEvents, err := db.QueryEvents(filters, db.GetClient(), "grain")
 		if err != nil {
-			fmt.Println("Error sending event:", err)
+			fmt.Println("Error querying events:", err)
-			response.SendClosed(ws, subID, "error: could not send event")
+			response.SendClosed(ws, subID, "error: could not query events")
 			return
 		}
-	}
 
-	// Indicate end of stored events
+		// Log the number of events retrieved
-	eoseMsg := []interface{}{"EOSE", subID}
+		fmt.Printf("Retrieved %d events for subscription %s\n", len(queriedEvents), subID)
-	eoseBytes, _ := json.Marshal(eoseMsg)
+		if len(queriedEvents) == 0 {
-	err = websocket.Message.Send(ws, string(eoseBytes))
+			fmt.Printf("No matching events found for subscription %s\n", subID)
-	if err != nil {
+		}
-		fmt.Println("Error sending EOSE:", err)
-		response.SendClosed(ws, subID, "error: could not send EOSE")
-		return
-	}
 
-	fmt.Println("Subscription handling completed, keeping connection open.")
+		// Send each event back to the client
-}
+		for _, evt := range queriedEvents {
-
+			msg := []interface{}{"EVENT", subID, evt}
-// Initialize the worker pool when your server starts
+			msgBytes, _ := json.Marshal(msg)
-func init() {
+			err = websocket.Message.Send(ws, string(msgBytes))
-	StartWorkerPool(10) // Adjust the number of workers as needed
+			if err != nil {
+				fmt.Println("Error sending event:", err)
+				response.SendClosed(ws, subID, "error: could not send event")
+				return
+			}
+		}
+
+		// Indicate end of stored events
+		eoseMsg := []interface{}{"EOSE", subID}
+		eoseBytes, _ := json.Marshal(eoseMsg)
+		err = websocket.Message.Send(ws, string(eoseBytes))
+		if err != nil {
+			fmt.Println("Error sending EOSE:", err)
+			response.SendClosed(ws, subID, "error: could not send EOSE")
+			return
+		}
+
+		fmt.Println("Subscription handling completed, keeping connection open.")
+	})
 }