main.go raw

   1  // Command sentry watches a Nostr relay for kind-1 text notes,
   2  // runs each through the dendrite 8-pass lattice detection chain,
   3  // and broadcasts verdict replies to every known relay from a fresh
   4  // throwaway npub.
   5  //
   6  // Relay discovery: subscribes to kind-10002 (NIP-65) events to
   7  // continuously harvest relay URLs. Verdicts are blasted to every
   8  // relay that will accept them.
   9  //
  10  // Each verdict is signed by a unique, single-use keypair — there is
  11  // no persistent bot identity to mute or flag.
  12  //
  13  // Usage:
  14  //
  15  //	sentry --watch wss://relay.orly.dev --memory .recognise_db
  16  package main
  17  
  18  import (
  19  	"context"
  20  	"flag"
  21  	"log"
  22  	"os"
  23  	"os/signal"
  24  	"strings"
  25  	"sync"
  26  	"time"
  27  
  28  	"git.mleku.dev/mleku/dendrite/pkg/nostr"
  29  )
  30  
  31  // sentrySignature is a string present in every verdict reply's content.
  32  // Events containing this string are our own verdicts — skip them to avoid
  33  // self-detection loops. Using content rather than tags because tags can
  34  // be spoofed by anyone to evade detection.
  35  const sentrySignature = "https://git.nostrdev.com/mleku/dendrite"
  36  
  37  func main() {
  38  	var (
  39  		watchURL      = flag.String("watch", "wss://relay.orly.dev", "relay to subscribe for events")
  40  		memoryDir     = flag.String("memory", ".recognise_db", "badger DB with trained mindsicles")
  41  		trollMemory   = flag.String("troll-memory", "", "badger DB with manipulation-trained mindsicles (empty = disabled)")
  42  		trollPasses   = flag.Int("troll-passes", 8, "number of troll detection passes")
  43  		trollThreshold = flag.Float64("troll-threshold", 0.05, "minimum troll score to include in verdict")
  44  		passes        = flag.Int("passes", 8, "number of detection passes")
  45  		window        = flag.Int("window", 500, "max tokens per sample")
  46  		workers       = flag.Int("workers", 4, "concurrent detector workers")
  47  		rateLimit     = flag.Int("rate-limit", 6, "max verdicts per minute")
  48  		minContent    = flag.Int("min-content", 50, "minimum content length to analyze")
  49  		threshold     = flag.Float64("threshold", 0.5, "minimum confidence to publish verdict")
  50  		maxRelays     = flag.Int("max-relays", 1000, "max relay URLs to discover")
  51  		verbose       = flag.Bool("verbose", false, "log detection details")
  52  	)
  53  	flag.Parse()
  54  
  55  	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
  56  	defer cancel()
  57  
  58  	// Load trained lattices.
  59  	log.Printf("loading %d-pass lattice chain from %s", *passes, *memoryDir)
  60  	detector, err := NewDetector(*memoryDir, *passes, *window)
  61  	if err != nil {
  62  		log.Fatalf("init detector: %v", err)
  63  	}
  64  	log.Printf("lattices loaded (%d passes, %d token window)", *passes, *window)
  65  
  66  	// Load troll detection lattices if configured.
  67  	if *trollMemory != "" {
  68  		log.Printf("loading %d-pass troll lattice chain from %s", *trollPasses, *trollMemory)
  69  		if err := detector.LoadTrollLattices(*trollMemory, *trollPasses); err != nil {
  70  			log.Fatalf("init troll detector: %v", err)
  71  		}
  72  		log.Printf("troll lattices loaded (%d passes)", *trollPasses)
  73  	}
  74  
  75  	// Relay pool: seeds + discovered from kind-10002 events.
  76  	pool := NewRelayPool(seedRelays, *maxRelays)
  77  	log.Printf("relay pool seeded with %d relays", pool.Size())
  78  
  79  	// Upload binary to blossom servers for content-addressable distribution.
  80  	log.Println("uploading sentry binary to blossom servers...")
  81  	blob, err := UploadSelf(ctx)
  82  	if err != nil {
  83  		log.Printf("blossom upload failed (continuing without): %v", err)
  84  	}
  85  
  86  	// Broadcaster fans out to entire pool.
  87  	broadcaster := NewBroadcaster(pool, blob)
  88  
  89  	// Rate limiter.
  90  	limiter := NewRateLimiter(*rateLimit)
  91  	defer limiter.Stop()
  92  
  93  	// Work channel for detector workers.
  94  	work := make(chan *nostr.Event, 64)
  95  
  96  	// Start detector workers.
  97  	var wg sync.WaitGroup
  98  	for i := 0; i < *workers; i++ {
  99  		wg.Add(1)
 100  		go func(id int) {
 101  			defer wg.Done()
 102  			for ev := range work {
 103  				verdict := detector.Detect(ctx, ev.Content)
 104  				if *verbose {
 105  					label := verdict.Label
 106  					if verdict.TrollLabel != "" {
 107  						label += " | " + verdict.TrollLabel
 108  					}
 109  					log.Printf("worker %d: event %s → %s", id, ev.ID[:12], label)
 110  				}
 111  				aiTriggered := !verdict.Human && verdict.Confidence >= *threshold
 112  				trollTriggered := verdict.TrollScore >= *trollThreshold && verdict.TrollLabel != ""
 113  				if aiTriggered || trollTriggered {
 114  					if limiter.Allow() {
 115  						reason := verdict.Label
 116  						if trollTriggered {
 117  							reason += " | " + verdict.TrollLabel
 118  						}
 119  						log.Printf("detection: event %s by %s — %s",
 120  							ev.ID[:12], ev.Pubkey[:12], reason)
 121  						broadcaster.PublishVerdict(ctx, ev, verdict)
 122  					} else if *verbose {
 123  						log.Printf("rate limited: skipping verdict for %s", ev.ID[:12])
 124  					}
 125  				}
 126  			}
 127  		}(i)
 128  	}
 129  
 130  	// Watch relay with reconnection.
 131  	log.Printf("watching %s for kind-1 + kind-10002 events", *watchURL)
 132  	watchLoop(ctx, *watchURL, work, pool, *minContent, *verbose)
 133  
 134  	close(work)
 135  	wg.Wait()
 136  	log.Println("shutdown complete")
 137  }
 138  
 139  // watchLoop subscribes to kind-1 (text notes) and kind-10002 (relay lists)
 140  // on the watch relay. Text notes go to the work channel for detection.
 141  // Relay lists feed the relay pool for broadcast discovery.
 142  func watchLoop(ctx context.Context, url string, work chan<- *nostr.Event, pool *RelayPool, minContent int, verbose bool) {
 143  	backoff := time.Second
 144  	lastPoolLog := time.Time{}
 145  
 146  	for {
 147  		select {
 148  		case <-ctx.Done():
 149  			return
 150  		default:
 151  		}
 152  
 153  		client, err := nostr.Connect(ctx, url)
 154  		if err != nil {
 155  			log.Printf("connect %s: %v (retry in %v)", url, err, backoff)
 156  			select {
 157  			case <-time.After(backoff):
 158  			case <-ctx.Done():
 159  				return
 160  			}
 161  			backoff = min(backoff*2, time.Minute)
 162  			continue
 163  		}
 164  		backoff = time.Second
 165  		log.Printf("connected to %s", url)
 166  
 167  		// Subscribe to kind-1 (text) and kind-10002 (relay lists).
 168  		if err := client.Subscribe(ctx, "sentry", nostr.Filter{
 169  			Kinds: []int{1, 10002},
 170  		}); err != nil {
 171  			log.Printf("subscribe %s: %v", url, err)
 172  			client.Disconnect()
 173  			continue
 174  		}
 175  
 176  		listenDone := make(chan error, 1)
 177  		go func() {
 178  			listenDone <- client.Listen(ctx)
 179  		}()
 180  
 181  		func() {
 182  			for {
 183  				select {
 184  				case <-ctx.Done():
 185  					client.Disconnect()
 186  					return
 187  				case err := <-listenDone:
 188  					log.Printf("disconnected from %s: %v", url, err)
 189  					return
 190  				case ev := <-client.Events:
 191  					if ev == nil {
 192  						continue
 193  					}
 194  
 195  					// Relay list: feed to pool for discovery.
 196  					if ev.Kind == 10002 {
 197  						pool.Ingest(ev)
 198  						if time.Since(lastPoolLog) > 10*time.Second {
 199  							log.Printf("relay pool: %d relays discovered", pool.Size())
 200  							lastPoolLog = time.Now()
 201  						}
 202  						continue
 203  					}
 204  
 205  					// Text note: feed to detector workers.
 206  					// Skip our own verdict replies. We identify them
 207  					// by a signature string in the content rather than
 208  					// by tag, because tags can be spoofed by anyone.
 209  					if strings.Contains(ev.Content, sentrySignature) {
 210  						continue
 211  					}
 212  					if len(ev.Content) < minContent {
 213  						continue
 214  					}
 215  					select {
 216  					case work <- ev:
 217  					default:
 218  						if verbose {
 219  							log.Printf("backpressure: dropping %s", ev.ID[:12])
 220  						}
 221  					}
 222  				}
 223  			}
 224  		}()
 225  
 226  		client.Disconnect()
 227  		log.Printf("reconnecting to %s in %v", url, backoff)
 228  		select {
 229  		case <-time.After(backoff):
 230  		case <-ctx.Done():
 231  			return
 232  		}
 233  	}
 234  }
 235  
 236  func min(a, b time.Duration) time.Duration {
 237  	if a < b {
 238  		return a
 239  	}
 240  	return b
 241  }
 242