mirror of
https://github.com/0ceanSlim/grain.git
synced 2024-11-21 16:17:13 +00:00
event purging refactor
This commit is contained in:
parent
8491827b06
commit
7e5e3758c3
@ -1,15 +1,11 @@
|
|||||||
package config
|
package config
|
||||||
|
|
||||||
type EventPurgeConfig struct {
|
type EventPurgeConfig struct {
|
||||||
Enabled bool `yaml:"enabled"`
|
Enabled bool `yaml:"enabled"`
|
||||||
KeepDurationDays int `yaml:"keep_duration_days"`
|
KeepIntervalHours int `yaml:"keep_interval_hours"`
|
||||||
PurgeIntervalHours int `yaml:"purge_interval_hours"`
|
PurgeIntervalMinutes int `yaml:"purge_interval_minutes"`
|
||||||
PurgeByCategory map[string]bool `yaml:"purge_by_category"`
|
PurgeByCategory map[string]bool `yaml:"purge_by_category"`
|
||||||
PurgeByKind []KindPurgeRule `yaml:"purge_by_kind"`
|
PurgeByKindEnabled bool `yaml:"purge_by_kind_enabled"`
|
||||||
ExcludeWhitelisted bool `yaml:"exclude_whitelisted"`
|
KindsToPurge []int `yaml:"kinds_to_purge"`
|
||||||
}
|
ExcludeWhitelisted bool `yaml:"exclude_whitelisted"`
|
||||||
|
|
||||||
type KindPurgeRule struct {
|
|
||||||
Kind int `yaml:"kind"`
|
|
||||||
Enabled bool `yaml:"enabled"`
|
|
||||||
}
|
}
|
||||||
|
@ -5,10 +5,13 @@ import (
|
|||||||
"grain/config"
|
"grain/config"
|
||||||
types "grain/config/types"
|
types "grain/config/types"
|
||||||
nostr "grain/server/types"
|
nostr "grain/server/types"
|
||||||
|
"grain/server/utils"
|
||||||
"log"
|
"log"
|
||||||
|
"strconv"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"go.mongodb.org/mongo-driver/bson"
|
"go.mongodb.org/mongo-driver/bson"
|
||||||
|
"go.mongodb.org/mongo-driver/mongo"
|
||||||
)
|
)
|
||||||
|
|
||||||
// PurgeOldEvents removes old events based on the configuration and a list of whitelisted pubkeys.
|
// PurgeOldEvents removes old events based on the configuration and a list of whitelisted pubkeys.
|
||||||
@ -18,49 +21,77 @@ func PurgeOldEvents(cfg *types.EventPurgeConfig) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
client := GetClient()
|
client := GetClient()
|
||||||
collection := client.Database("grain").Collection("events")
|
cutoff := time.Now().Add(-time.Duration(cfg.KeepIntervalHours) * time.Hour).Unix()
|
||||||
|
var collectionsToPurge []string
|
||||||
|
|
||||||
// Calculate the cutoff time
|
// Determine collections to purge
|
||||||
cutoff := time.Now().AddDate(0, 0, -cfg.KeepDurationDays).Unix()
|
if cfg.PurgeByKindEnabled {
|
||||||
|
for _, kind := range cfg.KindsToPurge {
|
||||||
// Create the base filter for fetching old events
|
collectionsToPurge = append(collectionsToPurge, "event-kind"+strconv.Itoa(kind))
|
||||||
baseFilter := bson.M{
|
}
|
||||||
"created_at": bson.M{"$lt": cutoff}, // Filter for events older than the cutoff
|
} else {
|
||||||
|
// If `purge_by_kind_enabled` is false, add all potential event kinds or find dynamically
|
||||||
|
collectionsToPurge = getAllEventCollections(client)
|
||||||
}
|
}
|
||||||
|
|
||||||
cursor, err := collection.Find(context.TODO(), baseFilter)
|
for _, collectionName := range collectionsToPurge {
|
||||||
if err != nil {
|
collection := client.Database("grain").Collection(collectionName)
|
||||||
log.Printf("Error fetching old events for purging: %v", err)
|
baseFilter := bson.M{"created_at": bson.M{"$lt": cutoff}}
|
||||||
return
|
|
||||||
}
|
|
||||||
defer cursor.Close(context.TODO())
|
|
||||||
|
|
||||||
for cursor.Next(context.TODO()) {
|
cursor, err := collection.Find(context.TODO(), baseFilter)
|
||||||
var evt nostr.Event
|
|
||||||
if err := cursor.Decode(&evt); err != nil {
|
|
||||||
log.Printf("Error decoding event: %v", err)
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
// Check if the event's pubkey is whitelisted and skip purging if configured to do so
|
|
||||||
if cfg.ExcludeWhitelisted && config.IsPubKeyWhitelisted(evt.PubKey) {
|
|
||||||
log.Printf("Skipping purging for whitelisted event ID: %s, pubkey: %s", evt.ID, evt.PubKey)
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
// Proceed with deleting the event if it is not whitelisted
|
|
||||||
_, err := collection.DeleteOne(context.TODO(), bson.M{"id": evt.ID})
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("Error purging event ID %s: %v", evt.ID, err)
|
log.Printf("Error fetching old events for purging from %s: %v", collectionName, err)
|
||||||
} else {
|
continue
|
||||||
log.Printf("Purged event ID: %s", evt.ID)
|
}
|
||||||
|
defer cursor.Close(context.TODO())
|
||||||
|
|
||||||
|
for cursor.Next(context.TODO()) {
|
||||||
|
var evt nostr.Event
|
||||||
|
if err := cursor.Decode(&evt); err != nil {
|
||||||
|
log.Printf("Error decoding event from %s: %v", collectionName, err)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
// Skip if the pubkey is whitelisted
|
||||||
|
if cfg.ExcludeWhitelisted && config.IsPubKeyWhitelisted(evt.PubKey) {
|
||||||
|
log.Printf("Skipping purging for whitelisted event ID: %s, pubkey: %s", evt.ID, evt.PubKey)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
// Check if purging by category is enabled and if the event matches the allowed category
|
||||||
|
category := utils.DetermineEventCategory(evt.Kind)
|
||||||
|
if purge, exists := cfg.PurgeByCategory[category]; exists && purge {
|
||||||
|
_, err := collection.DeleteOne(context.TODO(), bson.M{"id": evt.ID})
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("Error purging event ID %s from %s: %v", evt.ID, collectionName, err)
|
||||||
|
} else {
|
||||||
|
log.Printf("Purged event ID: %s from %s", evt.ID, collectionName)
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// getAllEventCollections returns a list of all event collections if purging all kinds.
|
||||||
|
func getAllEventCollections(client *mongo.Client) []string {
|
||||||
|
var collections []string
|
||||||
|
collectionNames, err := client.Database("grain").ListCollectionNames(context.TODO(), bson.M{})
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("Error listing collection names: %v", err)
|
||||||
|
return collections
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, name := range collectionNames {
|
||||||
|
if len(name) > 10 && name[:10] == "event-kind" {
|
||||||
|
collections = append(collections, name)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return collections
|
||||||
|
}
|
||||||
|
|
||||||
// ScheduleEventPurging runs the event purging at a configurable interval.
|
// ScheduleEventPurging runs the event purging at a configurable interval.
|
||||||
func ScheduleEventPurging(cfg *types.ServerConfig) {
|
func ScheduleEventPurging(cfg *types.ServerConfig) {
|
||||||
purgeInterval := time.Duration(cfg.EventPurge.PurgeIntervalHours) * time.Hour
|
purgeInterval := time.Duration(cfg.EventPurge.PurgeIntervalMinutes) * time.Minute
|
||||||
ticker := time.NewTicker(purgeInterval)
|
ticker := time.NewTicker(purgeInterval)
|
||||||
defer ticker.Stop()
|
defer ticker.Stop()
|
||||||
|
|
||||||
|
@ -183,7 +183,7 @@ func handleBlacklistAndWhitelist(ws *websocket.Conn, evt nostr.Event) bool {
|
|||||||
func handleRateAndSizeLimits(ws *websocket.Conn, evt nostr.Event, eventSize int) bool {
|
func handleRateAndSizeLimits(ws *websocket.Conn, evt nostr.Event, eventSize int) bool {
|
||||||
rateLimiter := config.GetRateLimiter()
|
rateLimiter := config.GetRateLimiter()
|
||||||
sizeLimiter := config.GetSizeLimiter()
|
sizeLimiter := config.GetSizeLimiter()
|
||||||
category := determineCategory(evt.Kind)
|
category := utils.DetermineEventCategory(evt.Kind)
|
||||||
|
|
||||||
if allowed, msg := rateLimiter.AllowEvent(evt.Kind, category); !allowed {
|
if allowed, msg := rateLimiter.AllowEvent(evt.Kind, category); !allowed {
|
||||||
response.SendOK(ws, evt.ID, false, msg)
|
response.SendOK(ws, evt.ID, false, msg)
|
||||||
@ -197,20 +197,3 @@ func handleRateAndSizeLimits(ws *websocket.Conn, evt nostr.Event, eventSize int)
|
|||||||
|
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
func determineCategory(kind int) string {
|
|
||||||
switch {
|
|
||||||
case kind == 0, kind == 3, kind >= 10000 && kind < 20000:
|
|
||||||
return "replaceable"
|
|
||||||
case kind == 1, kind >= 4 && kind < 45, kind >= 1000 && kind < 10000:
|
|
||||||
return "regular"
|
|
||||||
case kind == 2:
|
|
||||||
return "deprecated"
|
|
||||||
case kind >= 20000 && kind < 30000:
|
|
||||||
return "ephemeral"
|
|
||||||
case kind >= 30000 && kind < 40000:
|
|
||||||
return "parameterized_replaceable"
|
|
||||||
default:
|
|
||||||
return "unknown"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
18
server/utils/determineEventCategory.go
Normal file
18
server/utils/determineEventCategory.go
Normal file
@ -0,0 +1,18 @@
|
|||||||
|
package utils
|
||||||
|
|
||||||
|
func DetermineEventCategory(kind int) string {
|
||||||
|
switch {
|
||||||
|
case kind == 0, kind == 3, kind >= 10000 && kind < 20000:
|
||||||
|
return "replaceable"
|
||||||
|
case kind == 1, kind >= 4 && kind < 45, kind >= 1000 && kind < 10000:
|
||||||
|
return "regular"
|
||||||
|
case kind == 2:
|
||||||
|
return "deprecated"
|
||||||
|
case kind >= 20000 && kind < 30000:
|
||||||
|
return "ephemeral"
|
||||||
|
case kind >= 30000 && kind < 40000:
|
||||||
|
return "parameterized_replaceable"
|
||||||
|
default:
|
||||||
|
return "unknown"
|
||||||
|
}
|
||||||
|
}
|
Loading…
Reference in New Issue
Block a user