Add curation ACL mode and complete graph query implementation (v0.47.0)
Some checks failed
Go / build-and-release (push) Has been cancelled
Some checks failed
Go / build-and-release (push) Has been cancelled
Curation Mode: - Three-tier publisher classification: Trusted, Blacklisted, Unclassified - Per-pubkey rate limiting (default 50/day) for unclassified users - IP flood protection (default 500/day) with automatic banning - Event kind allow-listing via categories, ranges, and custom kinds - Query filtering hides blacklisted pubkey events (admin/owner exempt) - Web UI for managing trusted/blacklisted pubkeys and configuration - NIP-86 API endpoints for all curation management operations Graph Query Extension: - Complete reference aggregation for Badger and Neo4j backends - E-tag graph backfill migration (v8) runs automatically on startup - Configuration options: ORLY_GRAPH_QUERIES_ENABLED, MAX_DEPTH, etc. - NIP-11 advertisement of graph query capabilities Files modified: - app/handle-nip86-curating.go: NIP-86 curation API handlers (new) - app/web/src/CurationView.svelte: Curation management UI (new) - app/web/src/kindCategories.js: Kind category definitions (new) - pkg/acl/curating.go: Curating ACL implementation (new) - pkg/database/curating-acl.go: Database layer for curation (new) - pkg/neo4j/graph-refs.go: Neo4j ref collection (new) - pkg/database/migrations.go: E-tag graph backfill migration - pkg/protocol/graph/executor.go: Reference aggregation support - app/handle-event.go: Curation config event processing - app/handle-req.go: Blacklist filtering for queries - docs/GRAPH_QUERIES_REMAINING_PLAN.md: Updated completion status 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
woikos
989
pkg/database/curating-acl.go
Normal file
989
pkg/database/curating-acl.go
Normal file
@@ -0,0 +1,989 @@
|
||||
//go:build !(js && wasm)
|
||||
|
||||
package database
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"github.com/dgraph-io/badger/v4"
|
||||
)
|
||||
|
||||
// CuratingConfig represents the configuration for curating ACL mode
|
||||
// This is parsed from a kind 30078 event with d-tag "curating-config"
|
||||
type CuratingConfig struct {
|
||||
DailyLimit int `json:"daily_limit"` // Max events per day for unclassified users
|
||||
IPDailyLimit int `json:"ip_daily_limit"` // Max events per day from a single IP (flood protection)
|
||||
FirstBanHours int `json:"first_ban_hours"` // IP ban duration for first offense
|
||||
SecondBanHours int `json:"second_ban_hours"` // IP ban duration for second+ offense
|
||||
AllowedKinds []int `json:"allowed_kinds"` // Explicit kind numbers
|
||||
AllowedRanges []string `json:"allowed_ranges"` // Kind ranges like "1000-1999"
|
||||
KindCategories []string `json:"kind_categories"` // Category IDs like "social", "dm"
|
||||
ConfigEventID string `json:"config_event_id"` // ID of the config event
|
||||
ConfigPubkey string `json:"config_pubkey"` // Pubkey that published config
|
||||
ConfiguredAt int64 `json:"configured_at"` // Timestamp of config event
|
||||
}
|
||||
|
||||
// TrustedPubkey represents an explicitly trusted publisher
|
||||
type TrustedPubkey struct {
|
||||
Pubkey string `json:"pubkey"`
|
||||
Note string `json:"note,omitempty"`
|
||||
Added time.Time `json:"added"`
|
||||
}
|
||||
|
||||
// BlacklistedPubkey represents a blacklisted publisher
|
||||
type BlacklistedPubkey struct {
|
||||
Pubkey string `json:"pubkey"`
|
||||
Reason string `json:"reason,omitempty"`
|
||||
Added time.Time `json:"added"`
|
||||
}
|
||||
|
||||
// PubkeyEventCount tracks daily event counts for rate limiting
|
||||
type PubkeyEventCount struct {
|
||||
Pubkey string `json:"pubkey"`
|
||||
Date string `json:"date"` // YYYY-MM-DD format
|
||||
Count int `json:"count"`
|
||||
LastEvent time.Time `json:"last_event"`
|
||||
}
|
||||
|
||||
// IPOffense tracks rate limit violations from IPs
|
||||
type IPOffense struct {
|
||||
IP string `json:"ip"`
|
||||
OffenseCount int `json:"offense_count"`
|
||||
PubkeysHit []string `json:"pubkeys_hit"` // Pubkeys that hit rate limit from this IP
|
||||
LastOffense time.Time `json:"last_offense"`
|
||||
}
|
||||
|
||||
// CuratingBlockedIP represents a temporarily blocked IP with expiration
|
||||
type CuratingBlockedIP struct {
|
||||
IP string `json:"ip"`
|
||||
Reason string `json:"reason"`
|
||||
ExpiresAt time.Time `json:"expires_at"`
|
||||
Added time.Time `json:"added"`
|
||||
}
|
||||
|
||||
// SpamEvent represents an event flagged as spam
|
||||
type SpamEvent struct {
|
||||
EventID string `json:"event_id"`
|
||||
Pubkey string `json:"pubkey"`
|
||||
Reason string `json:"reason,omitempty"`
|
||||
Added time.Time `json:"added"`
|
||||
}
|
||||
|
||||
// UnclassifiedUser represents a user who hasn't been trusted or blacklisted
|
||||
type UnclassifiedUser struct {
|
||||
Pubkey string `json:"pubkey"`
|
||||
EventCount int `json:"event_count"`
|
||||
LastEvent time.Time `json:"last_event"`
|
||||
}
|
||||
|
||||
// CuratingACL database operations
|
||||
type CuratingACL struct {
|
||||
*D
|
||||
}
|
||||
|
||||
// NewCuratingACL creates a new CuratingACL instance
|
||||
func NewCuratingACL(db *D) *CuratingACL {
|
||||
return &CuratingACL{D: db}
|
||||
}
|
||||
|
||||
// ==================== Configuration ====================
|
||||
|
||||
// SaveConfig saves the curating configuration
|
||||
func (c *CuratingACL) SaveConfig(config CuratingConfig) error {
|
||||
return c.Update(func(txn *badger.Txn) error {
|
||||
key := c.getConfigKey()
|
||||
data, err := json.Marshal(config)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return txn.Set(key, data)
|
||||
})
|
||||
}
|
||||
|
||||
// GetConfig returns the curating configuration
|
||||
func (c *CuratingACL) GetConfig() (CuratingConfig, error) {
|
||||
var config CuratingConfig
|
||||
err := c.View(func(txn *badger.Txn) error {
|
||||
key := c.getConfigKey()
|
||||
item, err := txn.Get(key)
|
||||
if err != nil {
|
||||
if err == badger.ErrKeyNotFound {
|
||||
return nil // Return empty config
|
||||
}
|
||||
return err
|
||||
}
|
||||
val, err := item.ValueCopy(nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return json.Unmarshal(val, &config)
|
||||
})
|
||||
return config, err
|
||||
}
|
||||
|
||||
// IsConfigured returns true if a configuration event has been set
|
||||
func (c *CuratingACL) IsConfigured() (bool, error) {
|
||||
config, err := c.GetConfig()
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return config.ConfigEventID != "", nil
|
||||
}
|
||||
|
||||
// ==================== Trusted Pubkeys ====================
|
||||
|
||||
// SaveTrustedPubkey saves a trusted pubkey to the database
|
||||
func (c *CuratingACL) SaveTrustedPubkey(pubkey string, note string) error {
|
||||
return c.Update(func(txn *badger.Txn) error {
|
||||
key := c.getTrustedPubkeyKey(pubkey)
|
||||
trusted := TrustedPubkey{
|
||||
Pubkey: pubkey,
|
||||
Note: note,
|
||||
Added: time.Now(),
|
||||
}
|
||||
data, err := json.Marshal(trusted)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return txn.Set(key, data)
|
||||
})
|
||||
}
|
||||
|
||||
// RemoveTrustedPubkey removes a trusted pubkey from the database
|
||||
func (c *CuratingACL) RemoveTrustedPubkey(pubkey string) error {
|
||||
return c.Update(func(txn *badger.Txn) error {
|
||||
key := c.getTrustedPubkeyKey(pubkey)
|
||||
return txn.Delete(key)
|
||||
})
|
||||
}
|
||||
|
||||
// ListTrustedPubkeys returns all trusted pubkeys
|
||||
func (c *CuratingACL) ListTrustedPubkeys() ([]TrustedPubkey, error) {
|
||||
var trusted []TrustedPubkey
|
||||
err := c.View(func(txn *badger.Txn) error {
|
||||
prefix := c.getTrustedPubkeyPrefix()
|
||||
it := txn.NewIterator(badger.IteratorOptions{Prefix: prefix})
|
||||
defer it.Close()
|
||||
|
||||
for it.Rewind(); it.Valid(); it.Next() {
|
||||
item := it.Item()
|
||||
val, err := item.ValueCopy(nil)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
var t TrustedPubkey
|
||||
if err := json.Unmarshal(val, &t); err != nil {
|
||||
continue
|
||||
}
|
||||
trusted = append(trusted, t)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return trusted, err
|
||||
}
|
||||
|
||||
// IsPubkeyTrusted checks if a pubkey is trusted
|
||||
func (c *CuratingACL) IsPubkeyTrusted(pubkey string) (bool, error) {
|
||||
var trusted bool
|
||||
err := c.View(func(txn *badger.Txn) error {
|
||||
key := c.getTrustedPubkeyKey(pubkey)
|
||||
_, err := txn.Get(key)
|
||||
if err == badger.ErrKeyNotFound {
|
||||
trusted = false
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
trusted = true
|
||||
return nil
|
||||
})
|
||||
return trusted, err
|
||||
}
|
||||
|
||||
// ==================== Blacklisted Pubkeys ====================
|
||||
|
||||
// SaveBlacklistedPubkey saves a blacklisted pubkey to the database
|
||||
func (c *CuratingACL) SaveBlacklistedPubkey(pubkey string, reason string) error {
|
||||
return c.Update(func(txn *badger.Txn) error {
|
||||
key := c.getBlacklistedPubkeyKey(pubkey)
|
||||
blacklisted := BlacklistedPubkey{
|
||||
Pubkey: pubkey,
|
||||
Reason: reason,
|
||||
Added: time.Now(),
|
||||
}
|
||||
data, err := json.Marshal(blacklisted)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return txn.Set(key, data)
|
||||
})
|
||||
}
|
||||
|
||||
// RemoveBlacklistedPubkey removes a blacklisted pubkey from the database
|
||||
func (c *CuratingACL) RemoveBlacklistedPubkey(pubkey string) error {
|
||||
return c.Update(func(txn *badger.Txn) error {
|
||||
key := c.getBlacklistedPubkeyKey(pubkey)
|
||||
return txn.Delete(key)
|
||||
})
|
||||
}
|
||||
|
||||
// ListBlacklistedPubkeys returns all blacklisted pubkeys
|
||||
func (c *CuratingACL) ListBlacklistedPubkeys() ([]BlacklistedPubkey, error) {
|
||||
var blacklisted []BlacklistedPubkey
|
||||
err := c.View(func(txn *badger.Txn) error {
|
||||
prefix := c.getBlacklistedPubkeyPrefix()
|
||||
it := txn.NewIterator(badger.IteratorOptions{Prefix: prefix})
|
||||
defer it.Close()
|
||||
|
||||
for it.Rewind(); it.Valid(); it.Next() {
|
||||
item := it.Item()
|
||||
val, err := item.ValueCopy(nil)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
var b BlacklistedPubkey
|
||||
if err := json.Unmarshal(val, &b); err != nil {
|
||||
continue
|
||||
}
|
||||
blacklisted = append(blacklisted, b)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return blacklisted, err
|
||||
}
|
||||
|
||||
// IsPubkeyBlacklisted checks if a pubkey is blacklisted
|
||||
func (c *CuratingACL) IsPubkeyBlacklisted(pubkey string) (bool, error) {
|
||||
var blacklisted bool
|
||||
err := c.View(func(txn *badger.Txn) error {
|
||||
key := c.getBlacklistedPubkeyKey(pubkey)
|
||||
_, err := txn.Get(key)
|
||||
if err == badger.ErrKeyNotFound {
|
||||
blacklisted = false
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
blacklisted = true
|
||||
return nil
|
||||
})
|
||||
return blacklisted, err
|
||||
}
|
||||
|
||||
// ==================== Event Counting ====================
|
||||
|
||||
// GetEventCount returns the event count for a pubkey on a specific date
|
||||
func (c *CuratingACL) GetEventCount(pubkey, date string) (int, error) {
|
||||
var count int
|
||||
err := c.View(func(txn *badger.Txn) error {
|
||||
key := c.getEventCountKey(pubkey, date)
|
||||
item, err := txn.Get(key)
|
||||
if err == badger.ErrKeyNotFound {
|
||||
count = 0
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
val, err := item.ValueCopy(nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var ec PubkeyEventCount
|
||||
if err := json.Unmarshal(val, &ec); err != nil {
|
||||
return err
|
||||
}
|
||||
count = ec.Count
|
||||
return nil
|
||||
})
|
||||
return count, err
|
||||
}
|
||||
|
||||
// IncrementEventCount increments and returns the new event count for a pubkey
|
||||
func (c *CuratingACL) IncrementEventCount(pubkey, date string) (int, error) {
|
||||
var newCount int
|
||||
err := c.Update(func(txn *badger.Txn) error {
|
||||
key := c.getEventCountKey(pubkey, date)
|
||||
var ec PubkeyEventCount
|
||||
|
||||
item, err := txn.Get(key)
|
||||
if err == badger.ErrKeyNotFound {
|
||||
ec = PubkeyEventCount{
|
||||
Pubkey: pubkey,
|
||||
Date: date,
|
||||
Count: 0,
|
||||
LastEvent: time.Now(),
|
||||
}
|
||||
} else if err != nil {
|
||||
return err
|
||||
} else {
|
||||
val, err := item.ValueCopy(nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := json.Unmarshal(val, &ec); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
ec.Count++
|
||||
ec.LastEvent = time.Now()
|
||||
newCount = ec.Count
|
||||
|
||||
data, err := json.Marshal(ec)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return txn.Set(key, data)
|
||||
})
|
||||
return newCount, err
|
||||
}
|
||||
|
||||
// CleanupOldEventCounts removes event counts older than the specified date
|
||||
func (c *CuratingACL) CleanupOldEventCounts(beforeDate string) error {
|
||||
return c.Update(func(txn *badger.Txn) error {
|
||||
prefix := c.getEventCountPrefix()
|
||||
it := txn.NewIterator(badger.IteratorOptions{Prefix: prefix})
|
||||
defer it.Close()
|
||||
|
||||
var keysToDelete [][]byte
|
||||
for it.Rewind(); it.Valid(); it.Next() {
|
||||
item := it.Item()
|
||||
val, err := item.ValueCopy(nil)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
var ec PubkeyEventCount
|
||||
if err := json.Unmarshal(val, &ec); err != nil {
|
||||
continue
|
||||
}
|
||||
if ec.Date < beforeDate {
|
||||
keysToDelete = append(keysToDelete, item.KeyCopy(nil))
|
||||
}
|
||||
}
|
||||
|
||||
for _, key := range keysToDelete {
|
||||
if err := txn.Delete(key); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// ==================== IP Event Counting ====================
|
||||
|
||||
// IPEventCount tracks events from an IP address per day (flood protection)
|
||||
type IPEventCount struct {
|
||||
IP string `json:"ip"`
|
||||
Date string `json:"date"`
|
||||
Count int `json:"count"`
|
||||
LastEvent time.Time `json:"last_event"`
|
||||
}
|
||||
|
||||
// GetIPEventCount returns the total event count for an IP on a specific date
|
||||
func (c *CuratingACL) GetIPEventCount(ip, date string) (int, error) {
|
||||
var count int
|
||||
err := c.View(func(txn *badger.Txn) error {
|
||||
key := c.getIPEventCountKey(ip, date)
|
||||
item, err := txn.Get(key)
|
||||
if err == badger.ErrKeyNotFound {
|
||||
count = 0
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
val, err := item.ValueCopy(nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var ec IPEventCount
|
||||
if err := json.Unmarshal(val, &ec); err != nil {
|
||||
return err
|
||||
}
|
||||
count = ec.Count
|
||||
return nil
|
||||
})
|
||||
return count, err
|
||||
}
|
||||
|
||||
// IncrementIPEventCount increments and returns the new event count for an IP
|
||||
func (c *CuratingACL) IncrementIPEventCount(ip, date string) (int, error) {
|
||||
var newCount int
|
||||
err := c.Update(func(txn *badger.Txn) error {
|
||||
key := c.getIPEventCountKey(ip, date)
|
||||
var ec IPEventCount
|
||||
|
||||
item, err := txn.Get(key)
|
||||
if err == badger.ErrKeyNotFound {
|
||||
ec = IPEventCount{
|
||||
IP: ip,
|
||||
Date: date,
|
||||
Count: 0,
|
||||
LastEvent: time.Now(),
|
||||
}
|
||||
} else if err != nil {
|
||||
return err
|
||||
} else {
|
||||
val, err := item.ValueCopy(nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := json.Unmarshal(val, &ec); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
ec.Count++
|
||||
ec.LastEvent = time.Now()
|
||||
newCount = ec.Count
|
||||
|
||||
data, err := json.Marshal(ec)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return txn.Set(key, data)
|
||||
})
|
||||
return newCount, err
|
||||
}
|
||||
|
||||
// CleanupOldIPEventCounts removes IP event counts older than the specified date
|
||||
func (c *CuratingACL) CleanupOldIPEventCounts(beforeDate string) error {
|
||||
return c.Update(func(txn *badger.Txn) error {
|
||||
prefix := c.getIPEventCountPrefix()
|
||||
it := txn.NewIterator(badger.IteratorOptions{Prefix: prefix})
|
||||
defer it.Close()
|
||||
|
||||
var keysToDelete [][]byte
|
||||
for it.Rewind(); it.Valid(); it.Next() {
|
||||
item := it.Item()
|
||||
val, err := item.ValueCopy(nil)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
var ec IPEventCount
|
||||
if err := json.Unmarshal(val, &ec); err != nil {
|
||||
continue
|
||||
}
|
||||
if ec.Date < beforeDate {
|
||||
keysToDelete = append(keysToDelete, item.KeyCopy(nil))
|
||||
}
|
||||
}
|
||||
|
||||
for _, key := range keysToDelete {
|
||||
if err := txn.Delete(key); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
func (c *CuratingACL) getIPEventCountKey(ip, date string) []byte {
|
||||
buf := new(bytes.Buffer)
|
||||
buf.WriteString("CURATING_ACL_IP_EVENT_COUNT_")
|
||||
buf.WriteString(ip)
|
||||
buf.WriteString("_")
|
||||
buf.WriteString(date)
|
||||
return buf.Bytes()
|
||||
}
|
||||
|
||||
func (c *CuratingACL) getIPEventCountPrefix() []byte {
|
||||
return []byte("CURATING_ACL_IP_EVENT_COUNT_")
|
||||
}
|
||||
|
||||
// ==================== IP Offense Tracking ====================
|
||||
|
||||
// GetIPOffense returns the offense record for an IP
|
||||
func (c *CuratingACL) GetIPOffense(ip string) (*IPOffense, error) {
|
||||
var offense *IPOffense
|
||||
err := c.View(func(txn *badger.Txn) error {
|
||||
key := c.getIPOffenseKey(ip)
|
||||
item, err := txn.Get(key)
|
||||
if err == badger.ErrKeyNotFound {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
val, err := item.ValueCopy(nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
offense = new(IPOffense)
|
||||
return json.Unmarshal(val, offense)
|
||||
})
|
||||
return offense, err
|
||||
}
|
||||
|
||||
// RecordIPOffense records a rate limit violation from an IP for a pubkey
|
||||
// Returns the new offense count
|
||||
func (c *CuratingACL) RecordIPOffense(ip, pubkey string) (int, error) {
|
||||
var newCount int
|
||||
err := c.Update(func(txn *badger.Txn) error {
|
||||
key := c.getIPOffenseKey(ip)
|
||||
var offense IPOffense
|
||||
|
||||
item, err := txn.Get(key)
|
||||
if err == badger.ErrKeyNotFound {
|
||||
offense = IPOffense{
|
||||
IP: ip,
|
||||
OffenseCount: 0,
|
||||
PubkeysHit: []string{},
|
||||
LastOffense: time.Now(),
|
||||
}
|
||||
} else if err != nil {
|
||||
return err
|
||||
} else {
|
||||
val, err := item.ValueCopy(nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := json.Unmarshal(val, &offense); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// Add pubkey if not already in list
|
||||
found := false
|
||||
for _, p := range offense.PubkeysHit {
|
||||
if p == pubkey {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
offense.PubkeysHit = append(offense.PubkeysHit, pubkey)
|
||||
offense.OffenseCount++
|
||||
}
|
||||
offense.LastOffense = time.Now()
|
||||
newCount = offense.OffenseCount
|
||||
|
||||
data, err := json.Marshal(offense)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return txn.Set(key, data)
|
||||
})
|
||||
return newCount, err
|
||||
}
|
||||
|
||||
// ==================== IP Blocking ====================
|
||||
|
||||
// BlockIP blocks an IP for a specified duration
|
||||
func (c *CuratingACL) BlockIP(ip string, duration time.Duration, reason string) error {
|
||||
return c.Update(func(txn *badger.Txn) error {
|
||||
key := c.getBlockedIPKey(ip)
|
||||
blocked := CuratingBlockedIP{
|
||||
IP: ip,
|
||||
Reason: reason,
|
||||
ExpiresAt: time.Now().Add(duration),
|
||||
Added: time.Now(),
|
||||
}
|
||||
data, err := json.Marshal(blocked)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return txn.Set(key, data)
|
||||
})
|
||||
}
|
||||
|
||||
// UnblockIP removes an IP from the blocked list
|
||||
func (c *CuratingACL) UnblockIP(ip string) error {
|
||||
return c.Update(func(txn *badger.Txn) error {
|
||||
key := c.getBlockedIPKey(ip)
|
||||
return txn.Delete(key)
|
||||
})
|
||||
}
|
||||
|
||||
// IsIPBlocked checks if an IP is blocked and returns expiration time
|
||||
func (c *CuratingACL) IsIPBlocked(ip string) (bool, time.Time, error) {
|
||||
var blocked bool
|
||||
var expiresAt time.Time
|
||||
err := c.View(func(txn *badger.Txn) error {
|
||||
key := c.getBlockedIPKey(ip)
|
||||
item, err := txn.Get(key)
|
||||
if err == badger.ErrKeyNotFound {
|
||||
blocked = false
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
val, err := item.ValueCopy(nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var b CuratingBlockedIP
|
||||
if err := json.Unmarshal(val, &b); err != nil {
|
||||
return err
|
||||
}
|
||||
if time.Now().After(b.ExpiresAt) {
|
||||
// Block has expired
|
||||
blocked = false
|
||||
return nil
|
||||
}
|
||||
blocked = true
|
||||
expiresAt = b.ExpiresAt
|
||||
return nil
|
||||
})
|
||||
return blocked, expiresAt, err
|
||||
}
|
||||
|
||||
// ListBlockedIPs returns all blocked IPs (including expired ones)
|
||||
func (c *CuratingACL) ListBlockedIPs() ([]CuratingBlockedIP, error) {
|
||||
var blocked []CuratingBlockedIP
|
||||
err := c.View(func(txn *badger.Txn) error {
|
||||
prefix := c.getBlockedIPPrefix()
|
||||
it := txn.NewIterator(badger.IteratorOptions{Prefix: prefix})
|
||||
defer it.Close()
|
||||
|
||||
for it.Rewind(); it.Valid(); it.Next() {
|
||||
item := it.Item()
|
||||
val, err := item.ValueCopy(nil)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
var b CuratingBlockedIP
|
||||
if err := json.Unmarshal(val, &b); err != nil {
|
||||
continue
|
||||
}
|
||||
blocked = append(blocked, b)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return blocked, err
|
||||
}
|
||||
|
||||
// CleanupExpiredIPBlocks removes expired IP blocks
|
||||
func (c *CuratingACL) CleanupExpiredIPBlocks() error {
|
||||
return c.Update(func(txn *badger.Txn) error {
|
||||
prefix := c.getBlockedIPPrefix()
|
||||
it := txn.NewIterator(badger.IteratorOptions{Prefix: prefix})
|
||||
defer it.Close()
|
||||
|
||||
now := time.Now()
|
||||
var keysToDelete [][]byte
|
||||
for it.Rewind(); it.Valid(); it.Next() {
|
||||
item := it.Item()
|
||||
val, err := item.ValueCopy(nil)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
var b CuratingBlockedIP
|
||||
if err := json.Unmarshal(val, &b); err != nil {
|
||||
continue
|
||||
}
|
||||
if now.After(b.ExpiresAt) {
|
||||
keysToDelete = append(keysToDelete, item.KeyCopy(nil))
|
||||
}
|
||||
}
|
||||
|
||||
for _, key := range keysToDelete {
|
||||
if err := txn.Delete(key); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// ==================== Spam Events ====================
|
||||
|
||||
// MarkEventAsSpam marks an event as spam
|
||||
func (c *CuratingACL) MarkEventAsSpam(eventID, pubkey, reason string) error {
|
||||
return c.Update(func(txn *badger.Txn) error {
|
||||
key := c.getSpamEventKey(eventID)
|
||||
spam := SpamEvent{
|
||||
EventID: eventID,
|
||||
Pubkey: pubkey,
|
||||
Reason: reason,
|
||||
Added: time.Now(),
|
||||
}
|
||||
data, err := json.Marshal(spam)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return txn.Set(key, data)
|
||||
})
|
||||
}
|
||||
|
||||
// UnmarkEventAsSpam removes the spam flag from an event
|
||||
func (c *CuratingACL) UnmarkEventAsSpam(eventID string) error {
|
||||
return c.Update(func(txn *badger.Txn) error {
|
||||
key := c.getSpamEventKey(eventID)
|
||||
return txn.Delete(key)
|
||||
})
|
||||
}
|
||||
|
||||
// IsEventSpam checks if an event is marked as spam
|
||||
func (c *CuratingACL) IsEventSpam(eventID string) (bool, error) {
|
||||
var spam bool
|
||||
err := c.View(func(txn *badger.Txn) error {
|
||||
key := c.getSpamEventKey(eventID)
|
||||
_, err := txn.Get(key)
|
||||
if err == badger.ErrKeyNotFound {
|
||||
spam = false
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
spam = true
|
||||
return nil
|
||||
})
|
||||
return spam, err
|
||||
}
|
||||
|
||||
// ListSpamEvents returns all spam events
|
||||
func (c *CuratingACL) ListSpamEvents() ([]SpamEvent, error) {
|
||||
var spam []SpamEvent
|
||||
err := c.View(func(txn *badger.Txn) error {
|
||||
prefix := c.getSpamEventPrefix()
|
||||
it := txn.NewIterator(badger.IteratorOptions{Prefix: prefix})
|
||||
defer it.Close()
|
||||
|
||||
for it.Rewind(); it.Valid(); it.Next() {
|
||||
item := it.Item()
|
||||
val, err := item.ValueCopy(nil)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
var s SpamEvent
|
||||
if err := json.Unmarshal(val, &s); err != nil {
|
||||
continue
|
||||
}
|
||||
spam = append(spam, s)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return spam, err
|
||||
}
|
||||
|
||||
// ==================== Unclassified Users ====================
|
||||
|
||||
// ListUnclassifiedUsers returns users who are neither trusted nor blacklisted
|
||||
// sorted by event count descending
|
||||
func (c *CuratingACL) ListUnclassifiedUsers(limit int) ([]UnclassifiedUser, error) {
|
||||
// First, get all trusted and blacklisted pubkeys to exclude
|
||||
trusted, err := c.ListTrustedPubkeys()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
blacklisted, err := c.ListBlacklistedPubkeys()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
excludeSet := make(map[string]struct{})
|
||||
for _, t := range trusted {
|
||||
excludeSet[t.Pubkey] = struct{}{}
|
||||
}
|
||||
for _, b := range blacklisted {
|
||||
excludeSet[b.Pubkey] = struct{}{}
|
||||
}
|
||||
|
||||
// Now iterate through event counts and aggregate by pubkey
|
||||
pubkeyCounts := make(map[string]*UnclassifiedUser)
|
||||
|
||||
err = c.View(func(txn *badger.Txn) error {
|
||||
prefix := c.getEventCountPrefix()
|
||||
it := txn.NewIterator(badger.IteratorOptions{Prefix: prefix})
|
||||
defer it.Close()
|
||||
|
||||
for it.Rewind(); it.Valid(); it.Next() {
|
||||
item := it.Item()
|
||||
val, err := item.ValueCopy(nil)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
var ec PubkeyEventCount
|
||||
if err := json.Unmarshal(val, &ec); err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
// Skip if trusted or blacklisted
|
||||
if _, excluded := excludeSet[ec.Pubkey]; excluded {
|
||||
continue
|
||||
}
|
||||
|
||||
if existing, ok := pubkeyCounts[ec.Pubkey]; ok {
|
||||
existing.EventCount += ec.Count
|
||||
if ec.LastEvent.After(existing.LastEvent) {
|
||||
existing.LastEvent = ec.LastEvent
|
||||
}
|
||||
} else {
|
||||
pubkeyCounts[ec.Pubkey] = &UnclassifiedUser{
|
||||
Pubkey: ec.Pubkey,
|
||||
EventCount: ec.Count,
|
||||
LastEvent: ec.LastEvent,
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Convert to slice and sort by event count descending
|
||||
var users []UnclassifiedUser
|
||||
for _, u := range pubkeyCounts {
|
||||
users = append(users, *u)
|
||||
}
|
||||
sort.Slice(users, func(i, j int) bool {
|
||||
return users[i].EventCount > users[j].EventCount
|
||||
})
|
||||
|
||||
// Apply limit
|
||||
if limit > 0 && len(users) > limit {
|
||||
users = users[:limit]
|
||||
}
|
||||
|
||||
return users, nil
|
||||
}
|
||||
|
||||
// ==================== Key Generation ====================
|
||||
|
||||
func (c *CuratingACL) getConfigKey() []byte {
|
||||
return []byte("CURATING_ACL_CONFIG")
|
||||
}
|
||||
|
||||
func (c *CuratingACL) getTrustedPubkeyKey(pubkey string) []byte {
|
||||
buf := new(bytes.Buffer)
|
||||
buf.WriteString("CURATING_ACL_TRUSTED_PUBKEY_")
|
||||
buf.WriteString(pubkey)
|
||||
return buf.Bytes()
|
||||
}
|
||||
|
||||
func (c *CuratingACL) getTrustedPubkeyPrefix() []byte {
|
||||
return []byte("CURATING_ACL_TRUSTED_PUBKEY_")
|
||||
}
|
||||
|
||||
func (c *CuratingACL) getBlacklistedPubkeyKey(pubkey string) []byte {
|
||||
buf := new(bytes.Buffer)
|
||||
buf.WriteString("CURATING_ACL_BLACKLISTED_PUBKEY_")
|
||||
buf.WriteString(pubkey)
|
||||
return buf.Bytes()
|
||||
}
|
||||
|
||||
func (c *CuratingACL) getBlacklistedPubkeyPrefix() []byte {
|
||||
return []byte("CURATING_ACL_BLACKLISTED_PUBKEY_")
|
||||
}
|
||||
|
||||
func (c *CuratingACL) getEventCountKey(pubkey, date string) []byte {
|
||||
buf := new(bytes.Buffer)
|
||||
buf.WriteString("CURATING_ACL_EVENT_COUNT_")
|
||||
buf.WriteString(pubkey)
|
||||
buf.WriteString("_")
|
||||
buf.WriteString(date)
|
||||
return buf.Bytes()
|
||||
}
|
||||
|
||||
func (c *CuratingACL) getEventCountPrefix() []byte {
|
||||
return []byte("CURATING_ACL_EVENT_COUNT_")
|
||||
}
|
||||
|
||||
func (c *CuratingACL) getIPOffenseKey(ip string) []byte {
|
||||
buf := new(bytes.Buffer)
|
||||
buf.WriteString("CURATING_ACL_IP_OFFENSE_")
|
||||
buf.WriteString(ip)
|
||||
return buf.Bytes()
|
||||
}
|
||||
|
||||
func (c *CuratingACL) getBlockedIPKey(ip string) []byte {
|
||||
buf := new(bytes.Buffer)
|
||||
buf.WriteString("CURATING_ACL_BLOCKED_IP_")
|
||||
buf.WriteString(ip)
|
||||
return buf.Bytes()
|
||||
}
|
||||
|
||||
func (c *CuratingACL) getBlockedIPPrefix() []byte {
|
||||
return []byte("CURATING_ACL_BLOCKED_IP_")
|
||||
}
|
||||
|
||||
func (c *CuratingACL) getSpamEventKey(eventID string) []byte {
|
||||
buf := new(bytes.Buffer)
|
||||
buf.WriteString("CURATING_ACL_SPAM_EVENT_")
|
||||
buf.WriteString(eventID)
|
||||
return buf.Bytes()
|
||||
}
|
||||
|
||||
func (c *CuratingACL) getSpamEventPrefix() []byte {
|
||||
return []byte("CURATING_ACL_SPAM_EVENT_")
|
||||
}
|
||||
|
||||
// ==================== Kind Checking Helpers ====================
|
||||
|
||||
// IsKindAllowed checks if an event kind is allowed based on config
|
||||
func (c *CuratingACL) IsKindAllowed(kind int, config *CuratingConfig) bool {
|
||||
if config == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
// Check explicit kinds
|
||||
for _, k := range config.AllowedKinds {
|
||||
if k == kind {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
// Check ranges
|
||||
for _, rangeStr := range config.AllowedRanges {
|
||||
if kindInRange(kind, rangeStr) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
// Check categories
|
||||
for _, cat := range config.KindCategories {
|
||||
if kindInCategory(kind, cat) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
// kindInRange checks if a kind is within a range string like "1000-1999"
|
||||
func kindInRange(kind int, rangeStr string) bool {
|
||||
var start, end int
|
||||
n, err := fmt.Sscanf(rangeStr, "%d-%d", &start, &end)
|
||||
if err != nil || n != 2 {
|
||||
return false
|
||||
}
|
||||
return kind >= start && kind <= end
|
||||
}
|
||||
|
||||
// kindInCategory checks if a kind belongs to a predefined category
|
||||
func kindInCategory(kind int, category string) bool {
|
||||
categories := map[string][]int{
|
||||
"social": {0, 1, 3, 6, 7, 10002},
|
||||
"dm": {4, 14, 1059},
|
||||
"longform": {30023, 30024},
|
||||
"media": {1063, 20, 21, 22},
|
||||
"marketplace": {30017, 30018, 30019, 30020, 1021, 1022},
|
||||
"groups_nip29": {9, 10, 11, 12, 9000, 9001, 9002, 39000, 39001, 39002},
|
||||
"groups_nip72": {34550, 1111, 4550},
|
||||
"lists": {10000, 10001, 10003, 30000, 30001, 30003},
|
||||
}
|
||||
|
||||
kinds, ok := categories[category]
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
|
||||
for _, k := range kinds {
|
||||
if k == kind {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -38,5 +38,27 @@ func (a *GraphAdapter) TraverseThread(seedEventID []byte, maxDepth int, directio
|
||||
return a.db.TraverseThread(seedEventID, maxDepth, direction)
|
||||
}
|
||||
|
||||
// CollectInboundRefs implements graph.GraphDatabase.
|
||||
// It collects events that reference items in the result.
|
||||
func (a *GraphAdapter) CollectInboundRefs(result graph.GraphResultI, depth int, kinds []uint16) error {
|
||||
// Type assert to get the concrete GraphResult
|
||||
graphResult, ok := result.(*GraphResult)
|
||||
if !ok {
|
||||
return nil // Can't collect refs if we don't have a GraphResult
|
||||
}
|
||||
return a.db.AddInboundRefsToResult(graphResult, depth, kinds)
|
||||
}
|
||||
|
||||
// CollectOutboundRefs implements graph.GraphDatabase.
|
||||
// It collects events referenced by items in the result.
|
||||
func (a *GraphAdapter) CollectOutboundRefs(result graph.GraphResultI, depth int, kinds []uint16) error {
|
||||
// Type assert to get the concrete GraphResult
|
||||
graphResult, ok := result.(*GraphResult)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
return a.db.AddOutboundRefsToResult(graphResult, depth, kinds)
|
||||
}
|
||||
|
||||
// Verify GraphAdapter implements graph.GraphDatabase
|
||||
var _ graph.GraphDatabase = (*GraphAdapter)(nil)
|
||||
|
||||
@@ -325,3 +325,13 @@ func (r *GraphResult) GetTotalPubkeys() int {
|
||||
func (r *GraphResult) GetTotalEvents() int {
|
||||
return r.TotalEvents
|
||||
}
|
||||
|
||||
// GetInboundRefs returns the InboundRefs map for external access.
|
||||
func (r *GraphResult) GetInboundRefs() map[uint16]map[string][]string {
|
||||
return r.InboundRefs
|
||||
}
|
||||
|
||||
// GetOutboundRefs returns the OutboundRefs map for external access.
|
||||
func (r *GraphResult) GetOutboundRefs() map[uint16]map[string][]string {
|
||||
return r.OutboundRefs
|
||||
}
|
||||
|
||||
@@ -13,12 +13,13 @@ import (
|
||||
"next.orly.dev/pkg/database/indexes"
|
||||
"next.orly.dev/pkg/database/indexes/types"
|
||||
"git.mleku.dev/mleku/nostr/encoders/event"
|
||||
"git.mleku.dev/mleku/nostr/encoders/hex"
|
||||
"git.mleku.dev/mleku/nostr/encoders/ints"
|
||||
"git.mleku.dev/mleku/nostr/encoders/kind"
|
||||
)
|
||||
|
||||
const (
|
||||
currentVersion uint32 = 7
|
||||
currentVersion uint32 = 8
|
||||
)
|
||||
|
||||
func (d *D) RunMigrations() {
|
||||
@@ -115,6 +116,14 @@ func (d *D) RunMigrations() {
|
||||
// bump to version 7
|
||||
_ = d.writeVersionTag(7)
|
||||
}
|
||||
if dbVersion < 8 {
|
||||
log.I.F("migrating to version 8...")
|
||||
// Backfill e-tag graph indexes (eeg/gee) for graph query support
|
||||
// This creates edges for all existing events with e-tags
|
||||
d.BackfillETagGraph()
|
||||
// bump to version 8
|
||||
_ = d.writeVersionTag(8)
|
||||
}
|
||||
}
|
||||
|
||||
// writeVersionTag writes a new version tag key to the database (no value)
|
||||
@@ -1079,3 +1088,183 @@ func (d *D) RebuildWordIndexesWithNormalization() {
|
||||
|
||||
log.I.F("word index rebuild with unicode normalization complete")
|
||||
}
|
||||
|
||||
// BackfillETagGraph populates e-tag graph indexes (eeg/gee) for all existing events.
|
||||
// This enables graph traversal queries for thread/reply discovery.
|
||||
//
|
||||
// The migration:
|
||||
// 1. Iterates all events in compact storage (cmp prefix)
|
||||
// 2. Extracts e-tags from each event
|
||||
// 3. For e-tags referencing events we have, creates bidirectional edges:
|
||||
// - eeg|source|target|kind|direction(out) - forward edge
|
||||
// - gee|target|kind|direction(in)|source - reverse edge
|
||||
//
|
||||
// This is idempotent: running multiple times won't create duplicate edges
|
||||
// (BadgerDB overwrites existing keys).
|
||||
func (d *D) BackfillETagGraph() {
|
||||
log.I.F("backfilling e-tag graph indexes for graph query support...")
|
||||
var err error
|
||||
|
||||
type ETagEdge struct {
|
||||
SourceSerial *types.Uint40
|
||||
TargetSerial *types.Uint40
|
||||
Kind *types.Uint16
|
||||
}
|
||||
|
||||
var edges []ETagEdge
|
||||
var processedEvents int
|
||||
var eventsWithETags int
|
||||
var skippedTargets int
|
||||
|
||||
// First pass: collect all e-tag edges from events
|
||||
if err = d.View(func(txn *badger.Txn) error {
|
||||
// Iterate compact events (cmp prefix)
|
||||
cmpPrf := new(bytes.Buffer)
|
||||
if err = indexes.CompactEventEnc(nil).MarshalWrite(cmpPrf); chk.E(err) {
|
||||
return err
|
||||
}
|
||||
|
||||
it := txn.NewIterator(badger.IteratorOptions{Prefix: cmpPrf.Bytes()})
|
||||
defer it.Close()
|
||||
|
||||
for it.Rewind(); it.Valid(); it.Next() {
|
||||
item := it.Item()
|
||||
key := item.KeyCopy(nil)
|
||||
|
||||
// Extract serial from key (prefix 3 bytes + serial 5 bytes)
|
||||
if len(key) < 8 {
|
||||
continue
|
||||
}
|
||||
sourceSerial := new(types.Uint40)
|
||||
if err = sourceSerial.UnmarshalRead(bytes.NewReader(key[3:8])); chk.E(err) {
|
||||
continue
|
||||
}
|
||||
|
||||
// Get event data
|
||||
var val []byte
|
||||
if val, err = item.ValueCopy(nil); chk.E(err) {
|
||||
continue
|
||||
}
|
||||
|
||||
// Decode the event
|
||||
// First get the event ID from serial (needed for compact format decoding)
|
||||
eventId, idErr := d.GetEventIdBySerial(sourceSerial)
|
||||
if idErr != nil {
|
||||
continue
|
||||
}
|
||||
resolver := NewDatabaseSerialResolver(d, d.serialCache)
|
||||
ev, decErr := UnmarshalCompactEvent(val, eventId, resolver)
|
||||
if decErr != nil || ev == nil {
|
||||
continue
|
||||
}
|
||||
processedEvents++
|
||||
|
||||
// Extract e-tags
|
||||
eTags := ev.Tags.GetAll([]byte("e"))
|
||||
if len(eTags) == 0 {
|
||||
continue
|
||||
}
|
||||
eventsWithETags++
|
||||
|
||||
eventKind := new(types.Uint16)
|
||||
eventKind.Set(ev.Kind)
|
||||
|
||||
for _, eTag := range eTags {
|
||||
if eTag.Len() < 2 {
|
||||
continue
|
||||
}
|
||||
|
||||
// Get event ID from e-tag
|
||||
var targetEventID []byte
|
||||
targetEventID, err = hex.Dec(string(eTag.ValueHex()))
|
||||
if err != nil || len(targetEventID) != 32 {
|
||||
continue
|
||||
}
|
||||
|
||||
// Look up target event's serial
|
||||
targetSerial, lookupErr := d.GetSerialById(targetEventID)
|
||||
if lookupErr != nil || targetSerial == nil {
|
||||
// Target event not in our database - skip
|
||||
skippedTargets++
|
||||
continue
|
||||
}
|
||||
|
||||
edges = append(edges, ETagEdge{
|
||||
SourceSerial: sourceSerial,
|
||||
TargetSerial: targetSerial,
|
||||
Kind: eventKind,
|
||||
})
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}); chk.E(err) {
|
||||
log.E.F("e-tag graph backfill: failed to collect edges: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
log.I.F("e-tag graph backfill: processed %d events, %d with e-tags, found %d edges to create (%d targets not found)",
|
||||
processedEvents, eventsWithETags, len(edges), skippedTargets)
|
||||
|
||||
if len(edges) == 0 {
|
||||
log.I.F("e-tag graph backfill: no edges to create")
|
||||
return
|
||||
}
|
||||
|
||||
// Sort edges for ordered writes (improves compaction)
|
||||
sort.Slice(edges, func(i, j int) bool {
|
||||
if edges[i].SourceSerial.Get() != edges[j].SourceSerial.Get() {
|
||||
return edges[i].SourceSerial.Get() < edges[j].SourceSerial.Get()
|
||||
}
|
||||
return edges[i].TargetSerial.Get() < edges[j].TargetSerial.Get()
|
||||
})
|
||||
|
||||
// Second pass: write edges in batches
|
||||
const batchSize = 1000
|
||||
var createdEdges int
|
||||
|
||||
for i := 0; i < len(edges); i += batchSize {
|
||||
end := i + batchSize
|
||||
if end > len(edges) {
|
||||
end = len(edges)
|
||||
}
|
||||
batch := edges[i:end]
|
||||
|
||||
if err = d.Update(func(txn *badger.Txn) error {
|
||||
for _, edge := range batch {
|
||||
// Create forward edge: eeg|source|target|kind|direction(out)
|
||||
directionOut := new(types.Letter)
|
||||
directionOut.Set(types.EdgeDirectionETagOut)
|
||||
keyBuf := new(bytes.Buffer)
|
||||
if err = indexes.EventEventGraphEnc(edge.SourceSerial, edge.TargetSerial, edge.Kind, directionOut).MarshalWrite(keyBuf); chk.E(err) {
|
||||
continue
|
||||
}
|
||||
if err = txn.Set(keyBuf.Bytes(), nil); chk.E(err) {
|
||||
continue
|
||||
}
|
||||
|
||||
// Create reverse edge: gee|target|kind|direction(in)|source
|
||||
directionIn := new(types.Letter)
|
||||
directionIn.Set(types.EdgeDirectionETagIn)
|
||||
keyBuf.Reset()
|
||||
if err = indexes.GraphEventEventEnc(edge.TargetSerial, edge.Kind, directionIn, edge.SourceSerial).MarshalWrite(keyBuf); chk.E(err) {
|
||||
continue
|
||||
}
|
||||
if err = txn.Set(keyBuf.Bytes(), nil); chk.E(err) {
|
||||
continue
|
||||
}
|
||||
|
||||
createdEdges++
|
||||
}
|
||||
return nil
|
||||
}); chk.E(err) {
|
||||
log.W.F("e-tag graph backfill: batch write failed: %v", err)
|
||||
continue
|
||||
}
|
||||
|
||||
if (i/batchSize)%10 == 0 && i > 0 {
|
||||
log.I.F("e-tag graph backfill progress: %d/%d edges created", i, len(edges))
|
||||
}
|
||||
}
|
||||
|
||||
log.I.F("e-tag graph backfill complete: created %d bidirectional edges", createdEdges)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user