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