Compare commits

..

No commits in common. "6737718f08dfc8c2057d8933b9c2bbf707432271" and "c8ed954a9dac9675fc671ad9e209f8121c7ae05b" have entirely different histories.

4 changed files with 108 additions and 111 deletions

View File

@ -39,15 +39,19 @@ rate_limit:
# Rate limits for different event categories # Rate limits for different event categories
category_limits: category_limits:
ephemeral: ephemeral:
kind: 0
limit: 100 # Events per second limit: 100 # Events per second
burst: 200 # Allowed burst burst: 200 # Allowed burst
parameterized_replaceable: parameterized_replaceable:
kind: 0
limit: 5 limit: 5
burst: 10 burst: 10
regular: regular:
kind: 0
limit: 25 limit: 25
burst: 50 burst: 50
replaceable: replaceable:
kind: 0
limit: 10 limit: 10
burst: 20 burst: 20

View File

@ -1,109 +0,0 @@
package db
import (
"context"
"fmt"
relay "grain/server/types"
"go.mongodb.org/mongo-driver/bson"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/options"
)
// QueryEvents queries events from the MongoDB collection(s) based on filters
func QueryEvents(filters []relay.Filter, client *mongo.Client, databaseName string) ([]relay.Event, error) {
var results []relay.Event
for _, filter := range filters {
filterBson := bson.M{}
// Construct the BSON query based on the filters
if len(filter.IDs) > 0 {
filterBson["id"] = bson.M{"$in": filter.IDs}
}
if len(filter.Authors) > 0 {
filterBson["pubkey"] = bson.M{"$in": filter.Authors}
}
if len(filter.Kinds) > 0 {
filterBson["kind"] = bson.M{"$in": filter.Kinds}
}
if filter.Tags != nil {
for key, values := range filter.Tags {
if len(values) > 0 {
filterBson["tags."+key] = bson.M{"$in": values}
}
}
}
if filter.Since != nil {
filterBson["created_at"] = bson.M{"$gte": *filter.Since}
}
if filter.Until != nil {
if filterBson["created_at"] == nil {
filterBson["created_at"] = bson.M{"$lte": *filter.Until}
} else {
filterBson["created_at"].(bson.M)["$lte"] = *filter.Until
}
}
opts := options.Find().SetSort(bson.D{{Key: "created_at", Value: -1}})
if filter.Limit != nil {
opts.SetLimit(int64(*filter.Limit))
}
// If no specific kinds are specified, query all collections in the database
if len(filter.Kinds) == 0 {
collections, err := client.Database(databaseName).ListCollectionNames(context.TODO(), bson.D{})
if err != nil {
return nil, fmt.Errorf("error listing collections: %v", err)
}
for _, collectionName := range collections {
fmt.Printf("Querying collection: %s with query: %v\n", collectionName, filterBson)
collection := client.Database(databaseName).Collection(collectionName)
cursor, err := collection.Find(context.TODO(), filterBson, opts)
if err != nil {
return nil, fmt.Errorf("error querying collection %s: %v", collectionName, err)
}
defer cursor.Close(context.TODO())
for cursor.Next(context.TODO()) {
var event relay.Event
if err := cursor.Decode(&event); err != nil {
return nil, fmt.Errorf("error decoding event from collection %s: %v", collectionName, err)
}
results = append(results, event)
}
if err := cursor.Err(); err != nil {
return nil, fmt.Errorf("cursor error in collection %s: %v", collectionName, err)
}
}
} else {
// Query specific collections based on kinds
for _, kind := range filter.Kinds {
collectionName := fmt.Sprintf("event-kind%d", kind)
fmt.Printf("Querying collection: %s with query: %v\n", collectionName, filterBson)
collection := client.Database(databaseName).Collection(collectionName)
cursor, err := collection.Find(context.TODO(), filterBson, opts)
if err != nil {
return nil, fmt.Errorf("error querying collection %s: %v", collectionName, err)
}
defer cursor.Close(context.TODO())
for cursor.Next(context.TODO()) {
var event relay.Event
if err := cursor.Decode(&event); err != nil {
return nil, fmt.Errorf("error decoding event from collection %s: %v", collectionName, err)
}
results = append(results, event)
}
if err := cursor.Err(); err != nil {
return nil, fmt.Errorf("cursor error in collection %s: %v", collectionName, err)
}
}
}
}
return results, nil
}

View File

@ -1,6 +1,7 @@
package handlers package handlers
import ( import (
"context"
"encoding/json" "encoding/json"
"fmt" "fmt"
"grain/config" "grain/config"
@ -10,6 +11,9 @@ import (
"grain/server/utils" "grain/server/utils"
"sync" "sync"
"go.mongodb.org/mongo-driver/bson"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/options"
"golang.org/x/net/websocket" "golang.org/x/net/websocket"
) )
@ -73,7 +77,7 @@ func HandleReq(ws *websocket.Conn, message []interface{}, subscriptions map[stri
fmt.Printf("Subscription updated: %s with %d filters\n", subID, len(filters)) fmt.Printf("Subscription updated: %s with %d filters\n", subID, len(filters))
// Query the database with filters and send back the results // Query the database with filters and send back the results
queriedEvents, err := db.QueryEvents(filters, db.GetClient(), "grain") queriedEvents, err := QueryEvents(filters, db.GetClient(), "grain")
if err != nil { if err != nil {
fmt.Println("Error querying events:", err) fmt.Println("Error querying events:", err)
response.SendClosed(ws, subID, "error: could not query events") response.SendClosed(ws, subID, "error: could not query events")
@ -111,3 +115,101 @@ func HandleReq(ws *websocket.Conn, message []interface{}, subscriptions map[stri
fmt.Println("Subscription handling completed, keeping connection open.") fmt.Println("Subscription handling completed, keeping connection open.")
}) })
} }
// QueryEvents queries events from the MongoDB collection(s) based on filters
func QueryEvents(filters []relay.Filter, client *mongo.Client, databaseName string) ([]relay.Event, error) {
var results []relay.Event
for _, filter := range filters {
filterBson := bson.M{}
// Construct the BSON query based on the filters
if len(filter.IDs) > 0 {
filterBson["id"] = bson.M{"$in": filter.IDs}
}
if len(filter.Authors) > 0 {
filterBson["pubkey"] = bson.M{"$in": filter.Authors}
}
if len(filter.Kinds) > 0 {
filterBson["kind"] = bson.M{"$in": filter.Kinds}
}
if filter.Tags != nil {
for key, values := range filter.Tags {
if len(values) > 0 {
filterBson["tags."+key] = bson.M{"$in": values}
}
}
}
if filter.Since != nil {
filterBson["created_at"] = bson.M{"$gte": *filter.Since}
}
if filter.Until != nil {
if filterBson["created_at"] == nil {
filterBson["created_at"] = bson.M{"$lte": *filter.Until}
} else {
filterBson["created_at"].(bson.M)["$lte"] = *filter.Until
}
}
opts := options.Find().SetSort(bson.D{{Key: "created_at", Value: -1}})
if filter.Limit != nil {
opts.SetLimit(int64(*filter.Limit))
}
// If no specific kinds are specified, query all collections in the database
if len(filter.Kinds) == 0 {
collections, err := client.Database(databaseName).ListCollectionNames(context.TODO(), bson.D{})
if err != nil {
return nil, fmt.Errorf("error listing collections: %v", err)
}
for _, collectionName := range collections {
fmt.Printf("Querying collection: %s with query: %v\n", collectionName, filterBson)
collection := client.Database(databaseName).Collection(collectionName)
cursor, err := collection.Find(context.TODO(), filterBson, opts)
if err != nil {
return nil, fmt.Errorf("error querying collection %s: %v", collectionName, err)
}
defer cursor.Close(context.TODO())
for cursor.Next(context.TODO()) {
var event relay.Event
if err := cursor.Decode(&event); err != nil {
return nil, fmt.Errorf("error decoding event from collection %s: %v", collectionName, err)
}
results = append(results, event)
}
if err := cursor.Err(); err != nil {
return nil, fmt.Errorf("cursor error in collection %s: %v", collectionName, err)
}
}
} else {
// Query specific collections based on kinds
for _, kind := range filter.Kinds {
collectionName := fmt.Sprintf("event-kind%d", kind)
fmt.Printf("Querying collection: %s with query: %v\n", collectionName, filterBson)
collection := client.Database(databaseName).Collection(collectionName)
cursor, err := collection.Find(context.TODO(), filterBson, opts)
if err != nil {
return nil, fmt.Errorf("error querying collection %s: %v", collectionName, err)
}
defer cursor.Close(context.TODO())
for cursor.Next(context.TODO()) {
var event relay.Event
if err := cursor.Decode(&event); err != nil {
return nil, fmt.Errorf("error decoding event from collection %s: %v", collectionName, err)
}
results = append(results, event)
}
if err := cursor.Err(); err != nil {
return nil, fmt.Errorf("cursor error in collection %s: %v", collectionName, err)
}
}
}
}
return results, nil
}