relay.go raw
1 // Package relay implements a minimal NIP-01 Nostr relay.
2 //
3 // The relay accepts WebSocket connections from clients, handles EVENT
4 // (store + broadcast), REQ (subscribe + replay), and CLOSE (unsubscribe)
5 // messages. Events are stored in memory and matched against subscription
6 // filters. This is Stage 6 of the developmental plan — the organism
7 // learns to serve, not just consume.
8 package relay
9
10 import (
11 "context"
12 "crypto/rand"
13 "encoding/hex"
14 "encoding/json"
15 "net"
16 "net/http"
17 "slices"
18 "strings"
19 "sync"
20
21 "git.mleku.dev/mleku/dendrite/pkg/nostr"
22 "github.com/coder/websocket"
23 )
24
25 // Relay is a NIP-01 relay server with in-memory event storage.
26 type Relay struct {
27 // Name is the relay's display name for NIP-11.
28 Name string
29
30 // RequireAuth enables NIP-42 authentication. When true, clients
31 // must authenticate before publishing events.
32 RequireAuth bool
33
34 // ContentFilter is an optional coherence filter that rejects events
35 // whose content doesn't bond to any known lattice structure.
36 ContentFilter *CoherenceFilter
37
38 mu sync.RWMutex
39 events map[string]*nostr.Event // id → event
40 order []string // insertion order for limit queries
41
42 subMu sync.RWMutex
43 subs map[*conn]map[string][]nostr.Filter // conn → subID → filters
44
45 srv *http.Server
46 ln net.Listener
47 }
48
49 // conn wraps a WebSocket connection with a write mutex.
50 type conn struct {
51 ws *websocket.Conn
52 mu sync.Mutex
53 challenge string // NIP-42 auth challenge
54 authedAs string // pubkey of authenticated user (empty if not authed)
55 }
56
57 // New creates a relay with empty storage.
58 func New(name string) *Relay {
59 return &Relay{
60 Name: name,
61 events: make(map[string]*nostr.Event),
62 subs: make(map[*conn]map[string][]nostr.Filter),
63 }
64 }
65
66 // EventCount returns the number of stored events.
67 func (r *Relay) EventCount() int {
68 r.mu.RLock()
69 defer r.mu.RUnlock()
70 return len(r.events)
71 }
72
73 // Listen starts the relay on the given address (e.g., "127.0.0.1:0").
74 // Returns the actual address (useful when port is 0).
75 func (r *Relay) Listen(addr string) (string, error) {
76 ln, err := net.Listen("tcp", addr)
77 if err != nil {
78 return "", err
79 }
80 r.ln = ln
81
82 mux := http.NewServeMux()
83 mux.HandleFunc("/", func(w http.ResponseWriter, req *http.Request) {
84 // NIP-11: relay information document.
85 if req.Header.Get("Accept") == "application/nostr+json" {
86 r.handleNIP11(w)
87 return
88 }
89 // WebSocket upgrade.
90 ws, err := websocket.Accept(w, req, &websocket.AcceptOptions{
91 OriginPatterns: []string{"*"},
92 })
93 if err != nil {
94 return
95 }
96 r.handleConn(req.Context(), ws)
97 })
98
99 r.srv = &http.Server{Handler: mux}
100 go r.srv.Serve(ln)
101
102 return ln.Addr().String(), nil
103 }
104
105 // Shutdown stops the relay gracefully.
106 func (r *Relay) Shutdown(ctx context.Context) error {
107 if r.srv != nil {
108 return r.srv.Shutdown(ctx)
109 }
110 return nil
111 }
112
113 // Addr returns the listener address, or "" if not listening.
114 func (r *Relay) Addr() string {
115 if r.ln != nil {
116 return r.ln.Addr().String()
117 }
118 return ""
119 }
120
121 // handleNIP11 serves the NIP-11 relay information document.
122 func (r *Relay) handleNIP11(w http.ResponseWriter) {
123 info := map[string]any{
124 "name": r.Name,
125 "description": "dendrite lattice relay",
126 "supported_nips": []int{1, 9, 11, 42},
127 "software": "dendrite",
128 "version": "0.1.0",
129 }
130 w.Header().Set("Content-Type", "application/nostr+json")
131 json.NewEncoder(w).Encode(info)
132 }
133
134 // handleConn manages one WebSocket client session.
135 func (r *Relay) handleConn(ctx context.Context, ws *websocket.Conn) {
136 c := &conn{ws: ws}
137 defer func() {
138 r.removeSubs(c)
139 ws.CloseNow()
140 }()
141
142 // NIP-42: send auth challenge if authentication is enabled.
143 if r.RequireAuth {
144 var b [16]byte
145 rand.Read(b[:])
146 c.challenge = hex.EncodeToString(b[:])
147 r.sendAuth(ctx, c, c.challenge)
148 }
149
150 for {
151 _, data, err := ws.Read(ctx)
152 if err != nil {
153 return
154 }
155
156 var envelope []json.RawMessage
157 if json.Unmarshal(data, &envelope) != nil || len(envelope) < 2 {
158 r.sendNotice(ctx, c, "invalid message")
159 continue
160 }
161
162 var msgType string
163 if json.Unmarshal(envelope[0], &msgType) != nil {
164 continue
165 }
166
167 switch msgType {
168 case "EVENT":
169 r.handleEvent(ctx, c, envelope)
170 case "REQ":
171 r.handleReq(ctx, c, envelope)
172 case "CLOSE":
173 r.handleClose(ctx, c, envelope)
174 case "AUTH":
175 r.handleAuth(ctx, c, envelope)
176 }
177 }
178 }
179
180 // handleEvent processes an incoming EVENT message.
181 func (r *Relay) handleEvent(ctx context.Context, c *conn, envelope []json.RawMessage) {
182 if len(envelope) < 2 {
183 return
184 }
185 var event nostr.Event
186 if err := json.Unmarshal(envelope[1], &event); err != nil {
187 r.sendOK(ctx, c, "", false, "invalid: bad JSON")
188 return
189 }
190
191 // NIP-42: require authentication before accepting events.
192 if r.RequireAuth && c.authedAs == "" {
193 r.sendOK(ctx, c, event.ID, false, "auth-required: must authenticate")
194 return
195 }
196
197 // Validate.
198 if !event.Valid() {
199 r.sendOK(ctx, c, event.ID, false, "invalid: bad id or signature")
200 return
201 }
202
203 // Coherence filter: reject events that don't bond to any known structure.
204 if r.ContentFilter != nil && !r.ContentFilter.Admits(&event) {
205 r.sendOK(ctx, c, event.ID, false, "blocked: content incoherent with relay structure")
206 return
207 }
208
209 // NIP-09: kind 5 deletion events remove referenced events by the same author.
210 if event.Kind == 5 {
211 r.handleDeletion(ctx, c, &event)
212 return
213 }
214
215 // Store.
216 r.mu.Lock()
217 _, exists := r.events[event.ID]
218 if !exists {
219 r.events[event.ID] = &event
220 r.order = append(r.order, event.ID)
221 }
222 r.mu.Unlock()
223
224 r.sendOK(ctx, c, event.ID, true, "")
225
226 // Broadcast to matching subscriptions (skip duplicate).
227 if !exists {
228 r.broadcast(ctx, &event)
229 }
230 }
231
232 // handleAuth processes a NIP-42 AUTH response from a client.
233 // The client sends ["AUTH", <signed kind 22242 event>] with the
234 // challenge in a "challenge" tag and the relay URL in a "relay" tag.
235 func (r *Relay) handleAuth(ctx context.Context, c *conn, envelope []json.RawMessage) {
236 if len(envelope) < 2 {
237 return
238 }
239 var event nostr.Event
240 if err := json.Unmarshal(envelope[1], &event); err != nil {
241 r.sendOK(ctx, c, "", false, "invalid: bad auth event JSON")
242 return
243 }
244
245 // Must be kind 22242.
246 if event.Kind != 22242 {
247 r.sendOK(ctx, c, event.ID, false, "invalid: auth event must be kind 22242")
248 return
249 }
250
251 // Must have a valid signature.
252 if !event.Valid() {
253 r.sendOK(ctx, c, event.ID, false, "invalid: bad auth event signature")
254 return
255 }
256
257 // Must contain the challenge we sent.
258 challengeMatch := false
259 for _, tag := range event.Tags {
260 if len(tag) >= 2 && tag[0] == "challenge" && tag[1] == c.challenge {
261 challengeMatch = true
262 break
263 }
264 }
265 if !challengeMatch {
266 r.sendOK(ctx, c, event.ID, false, "invalid: challenge mismatch")
267 return
268 }
269
270 c.authedAs = event.Pubkey
271 r.sendOK(ctx, c, event.ID, true, "")
272 }
273
274 // handleDeletion processes a NIP-09 kind 5 deletion event.
275 // Only events authored by the same pubkey are deleted.
276 func (r *Relay) handleDeletion(ctx context.Context, c *conn, del *nostr.Event) {
277 r.mu.Lock()
278 deleted := 0
279 for _, tag := range del.Tags {
280 if len(tag) >= 2 && tag[0] == "e" {
281 targetID := tag[1]
282 if existing, ok := r.events[targetID]; ok {
283 if existing.Pubkey == del.Pubkey {
284 delete(r.events, targetID)
285 deleted++
286 }
287 }
288 }
289 }
290 // Store the deletion event itself.
291 if _, exists := r.events[del.ID]; !exists {
292 r.events[del.ID] = del
293 r.order = append(r.order, del.ID)
294 }
295 r.mu.Unlock()
296
297 r.sendOK(ctx, c, del.ID, true, "")
298 }
299
300 // handleReq processes a REQ message: ["REQ", "<sub_id>", <filter>, ...]
301 func (r *Relay) handleReq(ctx context.Context, c *conn, envelope []json.RawMessage) {
302 if len(envelope) < 3 {
303 return
304 }
305 var subID string
306 if json.Unmarshal(envelope[1], &subID) != nil {
307 return
308 }
309
310 var filters []nostr.Filter
311 for _, raw := range envelope[2:] {
312 var f nostr.Filter
313 if json.Unmarshal(raw, &f) == nil {
314 // Parse tag filters from raw JSON.
315 var m map[string]json.RawMessage
316 if json.Unmarshal(raw, &m) == nil {
317 for k, v := range m {
318 if strings.HasPrefix(k, "#") {
319 var vals []string
320 if json.Unmarshal(v, &vals) == nil {
321 if f.Tags == nil {
322 f.Tags = make(map[string][]string)
323 }
324 f.Tags[k[1:]] = vals
325 }
326 }
327 }
328 }
329 filters = append(filters, f)
330 }
331 }
332
333 // Register subscription.
334 r.subMu.Lock()
335 if r.subs[c] == nil {
336 r.subs[c] = make(map[string][]nostr.Filter)
337 }
338 r.subs[c][subID] = filters
339 r.subMu.Unlock()
340
341 // Replay stored events matching the filters.
342 r.mu.RLock()
343 var matched []*nostr.Event
344 for _, id := range r.order {
345 ev := r.events[id]
346 if matchesAny(ev, filters) {
347 matched = append(matched, ev)
348 }
349 }
350 r.mu.RUnlock()
351
352 // Apply limit (from last filter that has one).
353 limit := len(matched)
354 for i := len(filters) - 1; i >= 0; i-- {
355 if filters[i].Limit != nil {
356 limit = *filters[i].Limit
357 break
358 }
359 }
360 if limit < len(matched) {
361 // Return the most recent (last inserted) events.
362 matched = matched[len(matched)-limit:]
363 }
364
365 for _, ev := range matched {
366 r.sendEvent(ctx, c, subID, ev)
367 }
368
369 // EOSE.
370 r.sendEOSE(ctx, c, subID)
371 }
372
373 // handleClose processes a CLOSE message: ["CLOSE", "<sub_id>"]
374 func (r *Relay) handleClose(ctx context.Context, c *conn, envelope []json.RawMessage) {
375 if len(envelope) < 2 {
376 return
377 }
378 var subID string
379 if json.Unmarshal(envelope[1], &subID) != nil {
380 return
381 }
382
383 r.subMu.Lock()
384 if subs, ok := r.subs[c]; ok {
385 delete(subs, subID)
386 }
387 r.subMu.Unlock()
388 }
389
390 // removeSubs cleans up all subscriptions for a disconnected client.
391 func (r *Relay) removeSubs(c *conn) {
392 r.subMu.Lock()
393 delete(r.subs, c)
394 r.subMu.Unlock()
395 }
396
397 // broadcast sends a new event to all subscriptions that match it.
398 // Pre-marshals per subID to avoid redundant JSON encoding.
399 func (r *Relay) broadcast(ctx context.Context, ev *nostr.Event) {
400 r.subMu.RLock()
401 defer r.subMu.RUnlock()
402
403 // Cache pre-marshaled messages per subID to avoid redundant encoding.
404 cache := make(map[string][]byte)
405
406 for c, subs := range r.subs {
407 for subID, filters := range subs {
408 if matchesAny(ev, filters) {
409 data, ok := cache[subID]
410 if !ok {
411 data, _ = json.Marshal([]any{"EVENT", subID, ev})
412 cache[subID] = data
413 }
414 c.mu.Lock()
415 c.ws.Write(ctx, websocket.MessageText, data)
416 c.mu.Unlock()
417 }
418 }
419 }
420 }
421
422 // matchesAny returns true if the event matches any of the filters.
423 func matchesAny(ev *nostr.Event, filters []nostr.Filter) bool {
424 for _, f := range filters {
425 if matches(ev, f) {
426 return true
427 }
428 }
429 return false
430 }
431
432 // matches evaluates a single filter against an event.
433 func matches(ev *nostr.Event, f nostr.Filter) bool {
434 if len(f.IDs) > 0 && !containsPrefix(f.IDs, ev.ID) {
435 return false
436 }
437 if len(f.Authors) > 0 && !containsPrefix(f.Authors, ev.Pubkey) {
438 return false
439 }
440 if len(f.Kinds) > 0 && !containsInt(f.Kinds, ev.Kind) {
441 return false
442 }
443 if f.Since != nil && ev.CreatedAt < *f.Since {
444 return false
445 }
446 if f.Until != nil && ev.CreatedAt > *f.Until {
447 return false
448 }
449 // Tag filters.
450 for tagName, values := range f.Tags {
451 if !eventHasTagValue(ev, tagName, values) {
452 return false
453 }
454 }
455 return true
456 }
457
458 // containsPrefix returns true if any element of haystack is a prefix of needle (or equal).
459 func containsPrefix(haystack []string, needle string) bool {
460 for _, h := range haystack {
461 if strings.HasPrefix(needle, h) {
462 return true
463 }
464 }
465 return false
466 }
467
468 func containsInt(haystack []int, needle int) bool {
469 return slices.Contains(haystack, needle)
470 }
471
472 // eventHasTagValue returns true if the event has a tag with the given name
473 // whose value matches any of the provided values.
474 func eventHasTagValue(ev *nostr.Event, tagName string, values []string) bool {
475 for _, tag := range ev.Tags {
476 if len(tag) >= 2 && tag[0] == tagName {
477 for _, v := range values {
478 if tag[1] == v {
479 return true
480 }
481 }
482 }
483 }
484 return false
485 }
486
487 // Wire message helpers.
488
489 func (r *Relay) sendOK(ctx context.Context, c *conn, eventID string, accepted bool, msg string) {
490 data, _ := json.Marshal([]any{"OK", eventID, accepted, msg})
491 c.mu.Lock()
492 defer c.mu.Unlock()
493 c.ws.Write(ctx, websocket.MessageText, data)
494 }
495
496 func (r *Relay) sendEvent(ctx context.Context, c *conn, subID string, ev *nostr.Event) {
497 data, _ := json.Marshal([]any{"EVENT", subID, ev})
498 c.mu.Lock()
499 defer c.mu.Unlock()
500 c.ws.Write(ctx, websocket.MessageText, data)
501 }
502
503 func (r *Relay) sendEOSE(ctx context.Context, c *conn, subID string) {
504 data, _ := json.Marshal([]string{"EOSE", subID})
505 c.mu.Lock()
506 defer c.mu.Unlock()
507 c.ws.Write(ctx, websocket.MessageText, data)
508 }
509
510 func (r *Relay) sendNotice(ctx context.Context, c *conn, msg string) {
511 data, _ := json.Marshal([]string{"NOTICE", msg})
512 c.mu.Lock()
513 defer c.mu.Unlock()
514 c.ws.Write(ctx, websocket.MessageText, data)
515 }
516
517 func (r *Relay) sendAuth(ctx context.Context, c *conn, challenge string) {
518 data, _ := json.Marshal([]string{"AUTH", challenge})
519 c.mu.Lock()
520 defer c.mu.Unlock()
521 c.ws.Write(ctx, websocket.MessageText, data)
522 }
523
524 // Stats returns relay operational statistics.
525 type Stats struct {
526 Events int
527 Connections int
528 Subscriptions int
529 }
530
531 // Stats returns current relay statistics.
532 func (r *Relay) Stats() Stats {
533 r.mu.RLock()
534 events := len(r.events)
535 r.mu.RUnlock()
536
537 r.subMu.RLock()
538 conns := len(r.subs)
539 subs := 0
540 for _, s := range r.subs {
541 subs += len(s)
542 }
543 r.subMu.RUnlock()
544
545 return Stats{
546 Events: events,
547 Connections: conns,
548 Subscriptions: subs,
549 }
550 }
551
552