net.go raw
1 // Package net is the network domain. It owns every piece of relay protocol
2 // state: connections, subscriptions, authentication and pending work. Reader
3 // workers hand it parsed wire input; it answers with outbound frames, and it
4 // is the only caller of the database, verifier and Blossom domains.
5 package net
6
7 import (
8 "crypto/rand"
9 "encoding/hex"
10 "errors"
11 "fmt"
12 "sort"
13 "strconv"
14 "strings"
15 "time"
16
17 "git.smesh.lol/torly/pkg/config"
18 "git.smesh.lol/torly/pkg/logx"
19 "git.smesh.lol/torly/pkg/nostr"
20 tinyspawn "git.smesh.lol/torly/pkg/tinyspawn"
21 "git.smesh.lol/torly/pkg/wire"
22 )
23
24 const (
25 // The database and Blossom domains are single event loops. The network
26 // domain must never block while handing them work: a blocking send can
27 // deadlock when both domains fill their response buffers at once.
28 maxDBOut = 8192
29 maxBlobOut = 1024
30 // A queued upload can be large. Cap the total bytes held in memory.
31 maxBlobOutBytes = 256 << 20
32 )
33
34 const (
35 maxSubIDLen = 64
36 // pollInterval bounds message latency for worker-to-network traffic.
37 pollInterval = 2 * time.Millisecond
38 // pingInterval keeps idle WebSocket connections alive. Browsers do not
39 // send their own pings, and the reader closes idle sockets after 150s.
40 pingInterval = 50 * time.Second
41 )
42
43 // WorkerLink is the pair of channels to one reader/writer worker pair.
44 type WorkerLink struct {
45 In tinyspawn.Duplex[wire.Inbound]
46 Out tinyspawn.Duplex[wire.Outbound]
47 }
48
49 // Data is the domain's startup configuration.
50 type Data struct {
51 Cfg config.Config
52 RelaySk [32]byte
53 }
54
55 // Chans are the channels bound at spawn.
56 type Chans struct {
57 Workers []WorkerLink
58 DBReq tinyspawn.Duplex[wire.DBRequest]
59 DBResp tinyspawn.Duplex[wire.DBResponse]
60 BlobReq tinyspawn.Duplex[wire.BlobRequest]
61 BlobResp tinyspawn.Duplex[wire.BlobResponse]
62 Verifiers []tinyspawn.Duplex[wire.VerifyRequest]
63 VerifyResp []tinyspawn.Duplex[wire.VerifyResponse]
64 Ready chan<- wire.Ready
65 Stop tinyspawn.Breaker
66 }
67
68 type connState struct {
69 id wire.ConnID
70 worker int
71 ws bool
72 authed bool
73 pub [32]byte
74 challenge string
75 subs map[string][]nostr.Filter
76 closed bool
77 }
78
79 const (
80 vkEvent uint8 = iota
81 vkAuth
82 )
83
84 type verifyCtx struct {
85 conn *connState
86 event *nostr.Event
87 kind uint8
88 }
89
90 const (
91 dbSave uint8 = iota
92 dbQuery
93 dbCount
94 dbGraph
95 )
96
97 type dbCtx struct {
98 conn *connState
99 sub string
100 kind uint8
101 event *nostr.Event
102 graph nostr.GraphQuery
103 }
104
105 type blobCtx struct {
106 conn wire.ConnID
107 worker int
108 }
109
110 type server struct {
111 data Data
112 ch Chans
113
114 conns map[wire.ConnID]*connState
115 pending map[wire.ReqID]*verifyCtx
116 dbCtx map[wire.ReqID]*dbCtx
117 blobCtx map[wire.ReqID]*blobCtx
118 queue []*verifyCtx
119 busy []bool
120 nextReq wire.ReqID
121
122 dbOut []wire.DBRequest
123 blobOut []wire.BlobRequest
124 blobOutBytes int
125 }
126
127 // Run is the domain entry point.
128 func Run(data Data, ch Chans) {
129 s := &server{
130 data: data,
131 ch: ch,
132 conns: make(map[wire.ConnID]*connState),
133 pending: make(map[wire.ReqID]*verifyCtx),
134 dbCtx: make(map[wire.ReqID]*dbCtx),
135 blobCtx: make(map[wire.ReqID]*blobCtx),
136 busy: make([]bool, len(ch.Verifiers)),
137 }
138 ch.Ready <- wire.Ready{OK: true}
139 ticker := time.NewTicker(pollInterval)
140 defer ticker.Stop()
141 ping := time.NewTicker(pingInterval)
142 defer ping.Stop()
143 for {
144 s.flush()
145 worked := s.drain()
146 s.dispatch()
147 if worked {
148 continue
149 }
150 select {
151 case <-ticker.C:
152 case <-ping.C:
153 s.pingAll()
154 case <-ch.Stop.Done():
155 return
156 }
157 }
158 }
159
160 func (s *server) drain() bool {
161 worked := false
162 for i := range s.ch.Workers {
163 for j := 0; j < 32; j++ {
164 select {
165 case m := <-s.ch.Workers[i].In.Recv():
166 s.handleInbound(i, m)
167 worked = true
168 default:
169 j = 32
170 }
171 }
172 }
173 for {
174 select {
175 case r := <-s.ch.DBResp.Recv():
176 s.handleDB(r)
177 worked = true
178 default:
179 goto verifyDrain
180 }
181 }
182 verifyDrain:
183 for i := range s.ch.VerifyResp {
184 drained := false
185 for !drained {
186 select {
187 case r := <-s.ch.VerifyResp[i].Recv():
188 s.busy[i] = false
189 s.handleVerify(r)
190 worked = true
191 default:
192 drained = true
193 }
194 }
195 }
196
197 for {
198 select {
199 case r := <-s.ch.BlobResp.Recv():
200 s.handleBlob(r)
201 worked = true
202 default:
203 return worked
204 }
205 }
206 }
207
208 func (s *server) handleInbound(worker int, m wire.Inbound) {
209 switch m.Kind {
210 case wire.InOpened:
211 if _, ok := s.conns[m.Conn]; ok {
212 return
213 }
214 challenge := newChallenge()
215 c := &connState{
216 id: m.Conn,
217 worker: worker,
218 ws: true,
219 challenge: challenge,
220 subs: make(map[string][]nostr.Filter),
221 }
222 s.conns[m.Conn] = c
223 s.sendText(worker, m.Conn, authMsg(challenge))
224 case wire.InText:
225 c := s.conns[m.Conn]
226 if c == nil || c.closed {
227 return
228 }
229 s.handleMessage(c, m.Text)
230 case wire.InHTTP:
231 s.handleHTTP(worker, m.Conn, m.HTTP)
232 case wire.InClosed:
233 if c := s.conns[m.Conn]; c != nil {
234 c.closed = true
235 delete(s.conns, m.Conn)
236 }
237 }
238 }
239
240 func (s *server) handleHTTP(worker int, id wire.ConnID, req wire.HTTPRequest) {
241 if !s.data.Cfg.BlossomEnabled && req.Path != "/health" {
242 body := []byte("not found\n")
243 s.sendHTTP(worker, id, wire.HTTPResponse{
244 Status: 404,
245 Header: [][2]string{{"Content-Type", "text/plain"}, {"Content-Length", strconv.Itoa(len(body))}},
246 Body: body,
247 })
248 return
249 }
250 if req.Path == "/health" {
251 body := fmt.Sprintf(`{"status":"ok","connections":%d}`, len(s.conns))
252 s.sendHTTP(worker, id, wire.HTTPResponse{
253 Status: 200,
254 Header: [][2]string{{"Content-Type", "application/json"}, {"Content-Length", strconv.Itoa(len(body))}},
255 Body: []byte(body),
256 })
257 return
258 }
259 s.nextReq++
260 rid := s.nextReq
261 s.blobCtx[rid] = &blobCtx{conn: id, worker: worker}
262 s.sendBlob(wire.BlobRequest{
263 ID: rid,
264 Method: req.Method,
265 Path: req.Path,
266 Query: req.Query,
267 Header: req.Header,
268 Body: req.Body,
269 })
270 }
271
272 func (s *server) handleBlob(r wire.BlobResponse) {
273 ctx := s.blobCtx[r.ID]
274 if ctx == nil {
275 return
276 }
277 delete(s.blobCtx, r.ID)
278 if r.Handled {
279 s.sendHTTP(ctx.worker, ctx.conn, wire.HTTPResponse{Status: r.Status, Header: r.Header, Body: r.Body})
280 return
281 }
282 body := []byte("not found\n")
283 s.sendHTTP(ctx.worker, ctx.conn, wire.HTTPResponse{
284 Status: 404,
285 Header: [][2]string{{"Content-Type", "text/plain"}, {"Content-Length", strconv.Itoa(len(body))}},
286 Body: body,
287 })
288 }
289
290 // pingAll sends a WebSocket ping to every open subscription connection. A
291 // browser answers automatically, which resets the reader's idle deadline.
292 func (s *server) pingAll() {
293 for _, c := range s.conns {
294 if c.closed || !c.ws {
295 continue
296 }
297 s.send(c.worker, wire.Outbound{Conn: c.id, Kind: wire.OutPing})
298 }
299 }
300
301 func (s *server) handleMessage(c *connState, data []byte) {
302 parts, err := nostr.SplitArray(data)
303 if err != nil || len(parts) == 0 {
304 s.notice(c, "invalid: message is not a JSON array")
305 return
306 }
307 command, err := nostr.ParseJSONString(parts[0])
308 if err != nil {
309 s.notice(c, "invalid: command is not a string")
310 return
311 }
312 subID := func(raw []byte) (string, bool) {
313 v, err := nostr.ParseJSONString(raw)
314 if err != nil || len(v) == 0 || len(v) > maxSubIDLen {
315 return "", false
316 }
317 return v, true
318 }
319 switch command {
320 case "EVENT":
321 if len(parts) != 2 {
322 s.notice(c, "invalid: EVENT takes one event")
323 return
324 }
325 s.handleEvent(c, parts[1])
326 case "REQ":
327 if len(parts) < 3 {
328 s.notice(c, "invalid: REQ needs a subscription id and at least one filter")
329 return
330 }
331 sub, ok := subID(parts[1])
332 if !ok {
333 s.notice(c, "invalid: bad subscription id")
334 return
335 }
336 s.handleReq(c, sub, parts[2:])
337 case "CLOSE":
338 if len(parts) != 2 {
339 s.notice(c, "invalid: CLOSE takes a subscription id")
340 return
341 }
342 v, err := nostr.ParseJSONString(parts[1])
343 if err != nil {
344 s.notice(c, "invalid: bad subscription id")
345 return
346 }
347 delete(c.subs, v)
348 case "COUNT":
349 if len(parts) < 3 {
350 s.notice(c, "invalid: COUNT needs a subscription id and at least one filter")
351 return
352 }
353 sub, ok := subID(parts[1])
354 if !ok {
355 s.notice(c, "invalid: bad subscription id")
356 return
357 }
358 s.handleCount(c, sub, parts[2:])
359 case "AUTH":
360 if len(parts) != 2 {
361 s.notice(c, "invalid: AUTH takes one event")
362 return
363 }
364 s.handleAuth(c, parts[1])
365 default:
366 s.notice(c, "invalid: unknown command "+command)
367 }
368 }
369
370 func (s *server) handleEvent(c *connState, raw []byte) {
371 if len(raw) > s.data.Cfg.MaxEventBytes {
372 s.notice(c, "invalid: event is too large")
373 return
374 }
375 ev, err := nostr.ParseEvent(raw)
376 if err != nil {
377 if id := nostr.ExtractID(raw); id != "" {
378 s.ok(c, id, false, "invalid: "+err.Error())
379 } else {
380 s.notice(c, "invalid: "+err.Error())
381 }
382 return
383 }
384 idHex := nostr.Hex(ev.ID[:])
385 if err := s.validate(c, ev); err != nil {
386 s.ok(c, idHex, false, err.Error())
387 return
388 }
389 if !ev.CheckID() {
390 s.ok(c, idHex, false, "invalid: event id does not match contents")
391 return
392 }
393 s.enqueueVerify(c, ev, vkEvent)
394 }
395
396 func (s *server) handleAuth(c *connState, raw []byte) {
397 ev, err := nostr.ParseEvent(raw)
398 if err != nil {
399 s.notice(c, "invalid: "+err.Error())
400 return
401 }
402 idHex := nostr.Hex(ev.ID[:])
403 if err := s.validateAuth(c, ev); err != nil {
404 s.ok(c, idHex, false, "auth-required: "+err.Error())
405 return
406 }
407 s.enqueueVerify(c, ev, vkAuth)
408 }
409
410 func (s *server) validate(c *connState, ev *nostr.Event) error {
411 now := time.Now().Unix()
412 if ev.CreatedAt > now+s.data.Cfg.MaxFutureSecs {
413 return errors.New("invalid: created_at is too far in the future")
414 }
415 if ev.Expired(now) {
416 return errors.New("invalid: event has expired")
417 }
418 if len(ev.Tags) > 2000 {
419 return errors.New("invalid: too many tags")
420 }
421 if len(ev.Content) > s.data.Cfg.MaxEventBytes {
422 return errors.New("invalid: content is too large")
423 }
424 if s.data.Cfg.AuthRequired && !c.authed {
425 return errors.New("auth-required: authentication is required to publish")
426 }
427 if ev.IsProtected() && (!c.authed || c.pub != ev.PubKey) {
428 return errors.New("auth-required: this protected event may only be published by its author")
429 }
430 return nil
431 }
432
433 func (s *server) validateAuth(c *connState, ev *nostr.Event) error {
434 if ev.Kind != 22242 {
435 return errors.New("wrong event kind")
436 }
437 now := time.Now().Unix()
438 if ev.CreatedAt < now-600 || ev.CreatedAt > now+600 {
439 return errors.New("auth event timestamp is out of range")
440 }
441 challenge, ok := ev.Tags.First("challenge")
442 if !ok || challenge != c.challenge {
443 return errors.New("wrong challenge")
444 }
445 relayTag, ok := ev.Tags.First("relay")
446 if !ok {
447 return errors.New("missing relay tag")
448 }
449 if !relayURLMatches(relayTag, s.data.Cfg.RelayURL) {
450 return errors.New("relay URL does not match")
451 }
452 if !ev.CheckID() {
453 return errors.New("event id does not match contents")
454 }
455 return nil
456 }
457
458 func (s *server) enqueueVerify(c *connState, ev *nostr.Event, kind uint8) {
459 ctx := &verifyCtx{conn: c, event: ev, kind: kind}
460 s.queue = append(s.queue, ctx)
461 }
462
463 func (s *server) dispatch() {
464 for len(s.queue) > 0 {
465 slot := -1
466 for i := range s.busy {
467 if !s.busy[i] {
468 slot = i
469 break
470 }
471 }
472 if slot < 0 {
473 return
474 }
475 ctx := s.queue[0]
476 s.queue = s.queue[1:]
477 s.nextReq++
478 rid := s.nextReq
479 s.pending[rid] = ctx
480 s.busy[slot] = true
481 req := wire.VerifyRequest{ID: rid, Pub: ctx.event.PubKey, Msg: ctx.event.ID, Sig: ctx.event.Sig}
482 select {
483 case s.ch.Verifiers[slot].Tx <- req:
484 case <-s.ch.Stop.Done():
485 s.busy[slot] = false
486 return
487 }
488 }
489 }
490
491 func (s *server) handleVerify(r wire.VerifyResponse) {
492 ctx := s.pending[r.ID]
493 if ctx == nil {
494 return
495 }
496 delete(s.pending, r.ID)
497 if !r.OK {
498 idHex := nostr.Hex(ctx.event.ID[:])
499 if ctx.kind == vkAuth {
500 s.ok(ctx.conn, idHex, false, "auth-required: invalid signature")
501 } else {
502 s.ok(ctx.conn, idHex, false, "invalid: bad signature")
503 }
504 return
505 }
506 if ctx.kind == vkAuth {
507 ctx.conn.authed = true
508 ctx.conn.pub = ctx.event.PubKey
509 s.ok(ctx.conn, nostr.Hex(ctx.event.ID[:]), true, "")
510 return
511 }
512 s.accepted(ctx.conn, ctx.event)
513 }
514
515 func (s *server) accepted(c *connState, ev *nostr.Event) {
516 idHex := nostr.Hex(ev.ID[:])
517 if ev.Kind == 22242 {
518 s.ok(c, idHex, true, "")
519 return
520 }
521 if ev.IsEphemeral() {
522 s.ok(c, idHex, true, "")
523 s.broadcast(ev)
524 return
525 }
526 s.nextReq++
527 rid := s.nextReq
528 s.dbCtx[rid] = &dbCtx{conn: c, kind: dbSave, event: ev}
529 s.sendDB(wire.DBRequest{ID: rid, Kind: wire.DBSave, Event: ev.Clone()})
530 }
531
532 func (s *server) handleReq(c *connState, sub string, rawFilters [][]byte) {
533 filters, err := s.parseFilters(rawFilters)
534 if err != nil {
535 s.closed(c, sub, "invalid: "+err.Error())
536 return
537 }
538 if q := firstGraphQuery(filters); q != nil {
539 s.handleGraphReq(c, sub, *q)
540 return
541 }
542 if _, exists := c.subs[sub]; !exists && len(c.subs) >= s.data.Cfg.MaxSubsPerConn {
543 s.closed(c, sub, "error: too many subscriptions")
544 return
545 }
546 c.subs[sub] = filters
547 s.nextReq++
548 rid := s.nextReq
549 s.dbCtx[rid] = &dbCtx{conn: c, sub: sub, kind: dbQuery}
550 s.sendDB(wire.DBRequest{ID: rid, Kind: wire.DBQuery, Filters: cloneFilters(filters), Limit: s.data.Cfg.QueryLimit})
551 }
552
553 // firstGraphQuery returns the first _graph filter, matching orly: a graph
554 // request is answered once and the remaining filters are ignored.
555 func firstGraphQuery(filters []nostr.Filter) *nostr.GraphQuery {
556 for i := range filters {
557 if filters[i].Graph != nil {
558 return filters[i].Graph
559 }
560 }
561 return nil
562 }
563
564 func (s *server) handleGraphReq(c *connState, sub string, q nostr.GraphQuery) {
565 // A REQ with the same subscription id replaces the old one. A graph query
566 // is one-shot, so there is nothing to keep for live delivery.
567 delete(c.subs, sub)
568 if err := q.Validate(); err != nil {
569 s.closed(c, sub, "invalid: "+err.Error())
570 return
571 }
572 s.nextReq++
573 rid := s.nextReq
574 s.dbCtx[rid] = &dbCtx{conn: c, sub: sub, kind: dbGraph, graph: q}
575 s.sendDB(wire.DBRequest{ID: rid, Kind: wire.DBGraph, Graph: q, Limit: s.data.Cfg.QueryLimit})
576 }
577
578 func (s *server) handleCount(c *connState, sub string, rawFilters [][]byte) {
579 filters, err := s.parseFilters(rawFilters)
580 if err != nil {
581 s.closed(c, sub, "invalid: "+err.Error())
582 return
583 }
584 if firstGraphQuery(filters) != nil {
585 s.closed(c, sub, "error: _graph queries cannot be counted")
586 return
587 }
588 s.nextReq++
589 rid := s.nextReq
590 s.dbCtx[rid] = &dbCtx{conn: c, sub: sub, kind: dbCount}
591 s.sendDB(wire.DBRequest{ID: rid, Kind: wire.DBCount, Filters: cloneFilters(filters)})
592 }
593
594 // sendDB hands a request to the database domain without blocking the network
595 // domain. If the request channel is full the request waits in dbOut and is
596 // retried from Run. This keeps health and reads responsive while the database
597 // is busy and prevents the two event loops from deadlocking on full channels.
598 func (s *server) sendDB(req wire.DBRequest) {
599 if s.ch.DBReq.TrySend(req) {
600 return
601 }
602 if len(s.dbOut) >= maxDBOut {
603 if ctx := s.dbCtx[req.ID]; ctx != nil {
604 delete(s.dbCtx, req.ID)
605 logx.Errorf("db queue full: dropping request id=%d kind=%d", req.ID, req.Kind)
606 if ctx.conn != nil && !ctx.conn.closed {
607 if ctx.kind == dbSave && ctx.event != nil {
608 s.ok(ctx.conn, nostr.Hex(ctx.event.ID[:]), false, "error: relay busy")
609 } else {
610 s.closed(ctx.conn, ctx.sub, "error: relay busy")
611 }
612 }
613 }
614 return
615 }
616 s.dbOut = append(s.dbOut, req)
617 }
618
619 // sendBlob is the Blossom counterpart of sendDB. The byte cap bounds how much
620 // queued upload data the network domain can retain.
621 func (s *server) sendBlob(req wire.BlobRequest) {
622 if s.ch.BlobReq.TrySend(req) {
623 return
624 }
625 if len(s.blobOut) >= maxBlobOut || s.blobOutBytes+len(req.Body) > maxBlobOutBytes {
626 ctx := s.blobCtx[req.ID]
627 delete(s.blobCtx, req.ID)
628 logx.Errorf("blob queue full: dropping request id=%d bytes=%d", req.ID, len(req.Body))
629 if ctx != nil {
630 body := []byte("relay busy\n")
631 s.sendHTTP(ctx.worker, ctx.conn, wire.HTTPResponse{
632 Status: 503,
633 Header: [][2]string{{"Content-Type", "text/plain"}, {"Content-Length", strconv.Itoa(len(body))}},
634 Body: body,
635 })
636 }
637 return
638 }
639 s.blobOut = append(s.blobOut, req)
640 s.blobOutBytes += len(req.Body)
641 }
642
643 // flush retries queued database and Blossom requests. It never blocks.
644 func (s *server) flush() {
645 for len(s.dbOut) > 0 {
646 if !s.ch.DBReq.TrySend(s.dbOut[0]) {
647 break
648 }
649 s.dbOut[0] = wire.DBRequest{}
650 s.dbOut = s.dbOut[1:]
651 }
652 for len(s.blobOut) > 0 {
653 if !s.ch.BlobReq.TrySend(s.blobOut[0]) {
654 break
655 }
656 s.blobOutBytes -= len(s.blobOut[0].Body)
657 s.blobOut[0] = wire.BlobRequest{}
658 s.blobOut = s.blobOut[1:]
659 }
660 }
661
662 func (s *server) handleDB(r wire.DBResponse) {
663 ctx := s.dbCtx[r.ID]
664 if ctx == nil {
665 return
666 }
667 delete(s.dbCtx, r.ID)
668 if ctx.conn == nil || ctx.conn.closed {
669 return
670 }
671 switch ctx.kind {
672 case dbSave:
673 idHex := nostr.Hex(ctx.event.ID[:])
674 switch r.Code {
675 case wire.ResDuplicate:
676 s.ok(ctx.conn, idHex, true, "duplicate: already have this event")
677 case wire.ResReplaced:
678 s.ok(ctx.conn, idHex, false, "replaced: a newer event was already stored")
679 case wire.ResDeleted:
680 s.ok(ctx.conn, idHex, false, "deleted: this event was deleted")
681 case wire.ResOK:
682 s.ok(ctx.conn, idHex, true, "")
683 s.broadcast(ctx.event)
684 default:
685 s.ok(ctx.conn, idHex, false, "error: could not store event")
686 }
687 case dbQuery:
688 sortEvents(r.Events)
689 for i := range r.Events {
690 s.sendText(ctx.conn.worker, ctx.conn.id, eventMsg(ctx.sub, r.Events[i].AppendJSON(nil)))
691 }
692 s.sendText(ctx.conn.worker, ctx.conn.id, eoseMsg(ctx.sub))
693 case dbCount:
694 s.sendText(ctx.conn.worker, ctx.conn.id, countMsg(ctx.sub, r.Count, r.Approximate))
695 case dbGraph:
696 if r.Code == wire.ResError || r.Graph == nil {
697 msg := "error: graph query failed"
698 if r.Err != "" {
699 msg = "error: " + r.Err
700 }
701 s.closed(ctx.conn, ctx.sub, msg)
702 return
703 }
704 ev, err := s.graphEvent(ctx.graph, r.Graph)
705 if err != nil {
706 s.closed(ctx.conn, ctx.sub, "error: could not sign graph result")
707 return
708 }
709 s.sendText(ctx.conn.worker, ctx.conn.id, eventMsg(ctx.sub, ev.AppendJSON(nil)))
710 s.sendText(ctx.conn.worker, ctx.conn.id, eoseMsg(ctx.sub))
711 }
712 }
713
714 // graphEvent builds the relay-signed response event for a _graph query.
715 func (s *server) graphEvent(q nostr.GraphQuery, res *nostr.GraphResult) (*nostr.Event, error) {
716 ev := &nostr.Event{
717 Kind: nostr.KindGraphResult,
718 CreatedAt: time.Now().Unix(),
719 Tags: nostr.Tags{
720 {"edge", q.Edge},
721 {"direction", q.Direction},
722 {"seed", q.Seed},
723 {"depth", strconv.Itoa(q.Depth)},
724 {"elapsed", res.Elapsed},
725 },
726 Content: string(res.AppendContentJSON(nil)),
727 }
728 if err := ev.Sign(s.data.RelaySk); err != nil {
729 return nil, err
730 }
731 return ev, nil
732 }
733
734 func (s *server) broadcast(ev *nostr.Event) {
735 data := ev.AppendJSON(nil)
736 for _, c := range s.conns {
737 if c.closed {
738 continue
739 }
740 for sub, filters := range c.subs {
741 for i := range filters {
742 if filterMatches(ev, &filters[i]) {
743 s.sendText(c.worker, c.id, eventMsg(sub, data))
744 break
745 }
746 }
747 }
748 }
749 }
750
751 func (s *server) parseFilters(raw [][]byte) ([]nostr.Filter, error) {
752 if len(raw) == 0 {
753 return nil, errors.New("at least one filter is required")
754 }
755 if len(raw) > s.data.Cfg.MaxFilters {
756 return nil, errors.New("too many filters")
757 }
758 filters := make([]nostr.Filter, 0, len(raw))
759 for _, r := range raw {
760 f, err := nostr.ParseFilter(r)
761 if err != nil {
762 return nil, err
763 }
764 if f.Limit > s.data.Cfg.MaxLimit {
765 f.Limit = s.data.Cfg.MaxLimit
766 }
767 filters = append(filters, *f)
768 }
769 return filters, nil
770 }
771
772 func (s *server) sendText(worker int, id wire.ConnID, data []byte) {
773 s.send(worker, wire.Outbound{Conn: id, Kind: wire.OutText, Text: data})
774 }
775
776 func (s *server) sendHTTP(worker int, id wire.ConnID, resp wire.HTTPResponse) {
777 s.send(worker, wire.Outbound{Conn: id, Kind: wire.OutHTTP, HTTP: resp})
778 }
779
780 func (s *server) send(worker int, out wire.Outbound) {
781 if worker < 0 || worker >= len(s.ch.Workers) {
782 return
783 }
784 select {
785 case s.ch.Workers[worker].Out.Tx <- out:
786 case <-s.ch.Stop.Done():
787 default:
788 // A writer that cannot keep up loses the connection; its reader will
789 // observe the close and report it back.
790 }
791 }
792
793 func (s *server) ok(c *connState, id string, good bool, msg string) {
794 s.sendText(c.worker, c.id, okMsg(id, good, msg))
795 }
796
797 func (s *server) notice(c *connState, msg string) {
798 s.sendText(c.worker, c.id, noticeMsg(msg))
799 }
800
801 func (s *server) closed(c *connState, sub, msg string) {
802 s.sendText(c.worker, c.id, closedMsg(sub, msg))
803 }
804
805 func newChallenge() string {
806 b := make([]byte, 16)
807 if _, err := rand.Read(b); err != nil {
808 for i := range b {
809 b[i] = byte(i * 7)
810 }
811 }
812 return hex.EncodeToString(b)
813 }
814
815 func cloneFilters(in []nostr.Filter) []nostr.Filter {
816 out := make([]nostr.Filter, len(in))
817 for i := range in {
818 out[i] = in[i].Clone()
819 }
820 return out
821 }
822
823 func sortEvents(evs []nostr.Event) {
824 sort.SliceStable(evs, func(i, j int) bool {
825 if evs[i].CreatedAt != evs[j].CreatedAt {
826 return evs[i].CreatedAt > evs[j].CreatedAt
827 }
828 return string(evs[i].ID[:]) < string(evs[j].ID[:])
829 })
830 }
831
832 func relayURLMatches(a, b string) bool {
833 norm := func(s string) string {
834 s = strings.TrimSpace(s)
835 for _, p := range []string{"wss://", "ws://", "https://", "http://"} {
836 s = strings.TrimPrefix(s, p)
837 }
838 return strings.TrimSuffix(strings.ToLower(s), "/")
839 }
840 return a == b || norm(a) == norm(b)
841 }
842