2024-07-20 12:41:47 +00:00
|
|
|
package server
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
"encoding/json"
|
|
|
|
"fmt"
|
2024-07-23 14:53:01 +00:00
|
|
|
"grain/server/db"
|
2024-07-23 12:39:36 +00:00
|
|
|
"grain/server/events"
|
2024-07-23 00:51:44 +00:00
|
|
|
server "grain/server/types"
|
2024-07-23 12:39:36 +00:00
|
|
|
"grain/server/utils"
|
2024-07-22 17:50:43 +00:00
|
|
|
|
2024-07-20 12:41:47 +00:00
|
|
|
"golang.org/x/net/websocket"
|
|
|
|
)
|
|
|
|
|
2024-07-23 00:51:44 +00:00
|
|
|
var subscriptions = make(map[string]server.Subscription)
|
2024-07-20 12:41:47 +00:00
|
|
|
|
|
|
|
func Handler(ws *websocket.Conn) {
|
|
|
|
defer ws.Close()
|
|
|
|
|
|
|
|
var msg string
|
|
|
|
for {
|
|
|
|
err := websocket.Message.Receive(ws, &msg)
|
|
|
|
if err != nil {
|
|
|
|
fmt.Println("Error receiving message:", err)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
fmt.Println("Received message:", msg)
|
|
|
|
|
|
|
|
var message []interface{}
|
|
|
|
err = json.Unmarshal([]byte(msg), &message)
|
|
|
|
if err != nil {
|
|
|
|
fmt.Println("Error parsing message:", err)
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
if len(message) < 2 {
|
|
|
|
fmt.Println("Invalid message format")
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
messageType, ok := message[0].(string)
|
|
|
|
if !ok {
|
|
|
|
fmt.Println("Invalid message type")
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
switch messageType {
|
|
|
|
case "EVENT":
|
|
|
|
handleEvent(ws, message)
|
|
|
|
case "REQ":
|
|
|
|
handleReq(ws, message)
|
|
|
|
case "CLOSE":
|
|
|
|
handleClose(ws, message)
|
|
|
|
default:
|
|
|
|
fmt.Println("Unknown message type:", messageType)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func handleEvent(ws *websocket.Conn, message []interface{}) {
|
|
|
|
if len(message) != 2 {
|
|
|
|
fmt.Println("Invalid EVENT message format")
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
eventData, ok := message[1].(map[string]interface{})
|
|
|
|
if !ok {
|
|
|
|
fmt.Println("Invalid event data format")
|
|
|
|
return
|
|
|
|
}
|
|
|
|
eventBytes, err := json.Marshal(eventData)
|
|
|
|
if err != nil {
|
|
|
|
fmt.Println("Error marshaling event data:", err)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2024-07-23 14:53:01 +00:00
|
|
|
var evt server.Event
|
2024-07-20 12:41:47 +00:00
|
|
|
err = json.Unmarshal(eventBytes, &evt)
|
|
|
|
if err != nil {
|
|
|
|
fmt.Println("Error unmarshaling event data:", err)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2024-07-23 15:07:43 +00:00
|
|
|
// Call the HandleKind function
|
2024-07-23 14:53:01 +00:00
|
|
|
HandleKind(context.TODO(), evt, ws)
|
2024-07-20 12:41:47 +00:00
|
|
|
|
|
|
|
fmt.Println("Event processed:", evt.ID)
|
|
|
|
}
|
|
|
|
|
2024-07-23 14:53:01 +00:00
|
|
|
func HandleKind(ctx context.Context, evt server.Event, ws *websocket.Conn) {
|
|
|
|
if !utils.CheckSignature(evt) {
|
|
|
|
sendOKResponse(ws, evt.ID, false, "invalid: signature verification failed")
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
collection := db.GetCollection(evt.Kind)
|
|
|
|
|
|
|
|
var err error
|
|
|
|
switch evt.Kind {
|
|
|
|
case 0:
|
|
|
|
err = events.HandleKind0(ctx, evt, collection)
|
|
|
|
case 1:
|
|
|
|
err = events.HandleKind1(ctx, evt, collection)
|
|
|
|
default:
|
|
|
|
err = events.HandleUnknownKind(ctx, evt, collection)
|
|
|
|
}
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
sendOKResponse(ws, evt.ID, false, fmt.Sprintf("error: %v", err))
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
sendOKResponse(ws, evt.ID, true, "")
|
|
|
|
}
|
|
|
|
|
|
|
|
func sendOKResponse(ws *websocket.Conn, eventID string, status bool, message string) {
|
|
|
|
response := []interface{}{"OK", eventID, status, message}
|
|
|
|
responseBytes, _ := json.Marshal(response)
|
|
|
|
websocket.Message.Send(ws, string(responseBytes))
|
|
|
|
}
|
|
|
|
|
2024-07-20 12:41:47 +00:00
|
|
|
func handleReq(ws *websocket.Conn, message []interface{}) {
|
|
|
|
if len(message) < 3 {
|
|
|
|
fmt.Println("Invalid REQ message format")
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
subID, ok := message[1].(string)
|
|
|
|
if !ok {
|
|
|
|
fmt.Println("Invalid subscription ID format")
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2024-07-23 00:51:44 +00:00
|
|
|
filters := make([]server.Filter, len(message)-2)
|
2024-07-20 12:41:47 +00:00
|
|
|
for i, filter := range message[2:] {
|
|
|
|
filterData, ok := filter.(map[string]interface{})
|
|
|
|
if !ok {
|
|
|
|
fmt.Println("Invalid filter format")
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2024-07-23 00:51:44 +00:00
|
|
|
var f server.Filter
|
2024-07-22 17:50:43 +00:00
|
|
|
f.IDs = utils.ToStringArray(filterData["ids"])
|
|
|
|
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"])
|
2024-07-20 12:41:47 +00:00
|
|
|
|
|
|
|
filters[i] = f
|
|
|
|
}
|
|
|
|
|
2024-07-23 00:51:44 +00:00
|
|
|
subscriptions[subID] = server.Subscription{ID: subID, Filters: filters}
|
2024-07-20 12:41:47 +00:00
|
|
|
fmt.Println("Subscription added:", subID)
|
|
|
|
|
|
|
|
// Query the database with filters and send back the results
|
2024-07-23 15:07:43 +00:00
|
|
|
// TO DO why is this taking a certain kind as an argument for collection???
|
2024-07-23 14:53:01 +00:00
|
|
|
queriedEvents, err := QueryEvents(filters, db.GetClient(), "grain", "event-kind1")
|
2024-07-20 12:41:47 +00:00
|
|
|
if err != nil {
|
|
|
|
fmt.Println("Error querying events:", err)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2024-07-23 00:08:04 +00:00
|
|
|
for _, evt := range queriedEvents {
|
2024-07-20 12:41:47 +00:00
|
|
|
msg := []interface{}{"EVENT", subID, evt}
|
|
|
|
msgBytes, _ := json.Marshal(msg)
|
|
|
|
err = websocket.Message.Send(ws, string(msgBytes))
|
|
|
|
if err != nil {
|
|
|
|
fmt.Println("Error sending event:", err)
|
|
|
|
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)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func handleClose(ws *websocket.Conn, message []interface{}) {
|
|
|
|
if len(message) != 2 {
|
|
|
|
fmt.Println("Invalid CLOSE message format")
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
subID, ok := message[1].(string)
|
|
|
|
if !ok {
|
|
|
|
fmt.Println("Invalid subscription ID format")
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
delete(subscriptions, subID)
|
|
|
|
fmt.Println("Subscription closed:", subID)
|
|
|
|
|
|
|
|
closeMsg := []interface{}{"CLOSED", subID, "Subscription closed"}
|
|
|
|
closeBytes, _ := json.Marshal(closeMsg)
|
|
|
|
err := websocket.Message.Send(ws, string(closeBytes))
|
|
|
|
if err != nil {
|
|
|
|
fmt.Println("Error sending CLOSE message:", err)
|
|
|
|
return
|
|
|
|
}
|
2024-07-23 00:08:04 +00:00
|
|
|
}
|