tinyrelayfile
tinyrelay/internal/relay/relay.go/
internal/relay/relay.go raw
1package relay
2
3import (
4 "context"
5 "crypto/rand"
6 "encoding/hex"
7 "encoding/json"
8 "errors"
9 "fmt"
10 "net"
11 "net/http"
12 "net/url"
13 "strings"
14 "sync"
15 "time"
16
17 "github.com/FelineStateMachine/tinyrelay/internal/event"
18 "github.com/FelineStateMachine/tinyrelay/internal/syncprotocol"
19 "github.com/coder/websocket"
20)
21
22type Session struct {
23 // PubKeys contains public keys established by the connection's AUTH flow.
24 PubKeys []string
25 // RelayURL is the relay identity used when authenticating and publishing.
26 RelayURL string
27 // RemoteIP is the peer address observed by the HTTP server.
28 RemoteIP string
29}
30
31// Backend publishes while the Router's write gate is held. Publish performs
32// the backend's admission, persistence, duplicate handling or control action
33// before returning. A nil error allows Router to fan out the event unless the
34// returned reason identifies a duplicate. Persistence and delivery planning
35// remain backend responsibilities for event kinds that require them.
36type Backend interface {
37 Publish(ctx context.Context, event event.Event, session Session) (reason string, err error)
38 Query(ctx context.Context, filters []event.Filter, session Session) ([]event.Event, error)
39 Count(ctx context.Context, filters []event.Filter, session Session) (any, error)
40}
41
42// Reader-side privacy can be stricter than the query filter. It is checked for
43// both historical and live delivery, including encrypted signer traffic.
44type Reader interface {
45 CanRead(event event.Event, session Session) bool
46}
47
48// FilterReader applies reader privacy with the subscription filter available
49// to the policy decision.
50type FilterReader interface {
51 CanReadFilter(event event.Event, session Session, filter *event.Filter) bool
52}
53
54// QueryHints can return query results together with completion hints. Router
55// places these values in the EOSE response; the relay backend uses values such
56// as "finish", "more" and "auth".
57type QueryHints interface {
58 QueryHints(ctx context.Context, filters []event.Filter, session Session) ([]event.Event, []string, error)
59}
60
61// PolicyChanges can invalidate subscriptions in storage after an access
62// policy revision. Router removes client state before invoking the callback.
63type PolicyChanges interface {
64 CloseSubscriptions(session Session)
65}
66
67// SyncBackend supplies the authorized event snapshot for a NIP-77 session.
68// The Router holds its read gate while calling Sync, so a concurrent Publish
69// cannot create an opening between snapshot selection and NEG-OPEN.
70type SyncBackend interface {
71 Sync(ctx context.Context, filter event.Filter, session Session) ([]syncprotocol.Item, error)
72}
73
74type Config struct {
75 // RelayURL is the default relay identity included in sessions.
76 RelayURL string
77 // RequestRelayURL optionally derives the relay identity from each request.
78 RequestRelayURL func(*http.Request) string
79 // OnAuthenticate checks the authenticated session candidate and may persist
80 // authentication bookkeeping. The Session argument is a value snapshot;
81 // Router records the newly authenticated public key after this callback.
82 OnAuthenticate func(context.Context, Session) error
83 // MaxMessageBytes limits websocket text messages. Zero leaves the limit
84 // unlimited; a negative value selects the 512 KiB default.
85 MaxMessageBytes int64
86 // MaxPendingBytes bounds queued output per connection.
87 MaxPendingBytes int
88 // ReadLimit, WriteLimit and PingInterval control websocket liveness.
89 ReadLimit time.Duration
90 WriteLimit time.Duration
91 PingInterval time.Duration
92 // OriginPatterns is passed to the websocket origin check.
93 OriginPatterns []string
94 // OnConnection and OnSubscription receive +1 and -1 lifecycle deltas.
95 OnConnection func(int)
96 OnSubscription func(int)
97}
98
99func (c Config) withDefaults() Config {
100 if c.MaxMessageBytes < 0 {
101 c.MaxMessageBytes = 512 << 10
102 }
103 if c.MaxPendingBytes <= 0 {
104 c.MaxPendingBytes = 4 << 20
105 }
106 if c.ReadLimit <= 0 {
107 c.ReadLimit = 60 * time.Second
108 }
109 if c.WriteLimit <= 0 {
110 c.WriteLimit = 10 * time.Second
111 }
112 if c.PingInterval <= 0 {
113 c.PingInterval = 30 * time.Second
114 }
115 return c
116}
117
118type Router struct {
119 backend Backend
120 cfg Config
121 mu sync.RWMutex // publish is exclusive with REQ registration + query
122 clientsMu sync.Mutex
123 clients map[*client]struct{}
124 listenersMu sync.Mutex
125 listeners map[*listener]struct{}
126 closed chan struct{}
127 closeOnce sync.Once
128 clientsWG sync.WaitGroup
129}
130
131type listener struct{ fn func(event.Event) }
132
133func New(backend Backend, cfg Config) *Router {
134 return &Router{backend: backend, cfg: cfg.withDefaults(), clients: make(map[*client]struct{}), listeners: make(map[*listener]struct{}), closed: make(chan struct{})}
135}
136
137// Listen registers an in-process observer of live fan-out. It sees every
138// event delivered to websocket subscribers, before the reader's own access
139// check. The callback runs under the publish fence and must not block; the
140// returned function removes it.
141func (r *Router) Listen(fn func(event.Event)) func() {
142 l := &listener{fn: fn}
143 r.listenersMu.Lock()
144 r.listeners[l] = struct{}{}
145 r.listenersMu.Unlock()
146 return func() {
147 r.listenersMu.Lock()
148 delete(r.listeners, l)
149 r.listenersMu.Unlock()
150 }
151}
152
153func (r *Router) HandleHTTP(w http.ResponseWriter, req *http.Request) {
154 if req.Method != http.MethodGet {
155 http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
156 return
157 }
158 conn, err := websocket.Accept(w, req, &websocket.AcceptOptions{OriginPatterns: r.cfg.OriginPatterns})
159 if err != nil {
160 return
161 }
162 c := newClient(r, conn, req)
163 r.clientsMu.Lock()
164 select {
165 case <-r.closed:
166 r.clientsMu.Unlock()
167 _ = conn.Close(websocket.StatusGoingAway, "relay shutting down")
168 return
169 default:
170 }
171 r.clients[c] = struct{}{}
172 if r.cfg.OnConnection != nil {
173 r.cfg.OnConnection(1)
174 }
175 r.clientsWG.Add(1)
176 r.clientsMu.Unlock()
177 go c.run()
178}
179
180// Publish is the shared ingress path for websocket EVENT and HTTP-generated
181// events. It applies the same historical/live fence and fan-out rules.
182func (r *Router) Publish(ctx context.Context, e event.Event, s Session) (string, error) {
183 r.mu.Lock()
184 defer r.mu.Unlock()
185 reason, err := r.backend.Publish(ctx, e, s)
186 if err == nil && !duplicateReason(reason) {
187 if accessEvent(e) {
188 r.closeSubscriptionsLocked("blocked: relay access policy changed; subscribe again")
189 }
190 r.fanout(e)
191 }
192 return reason, err
193}
194
195func (r *Router) publishClient(c *client, e event.Event, s Session) (string, error) {
196 r.mu.Lock()
197 defer r.mu.Unlock()
198 reason, err := r.backend.Publish(c.ctx, e, s)
199 if err != nil {
200 return reason, err
201 }
202 // Queue the sender's OK before any live EVENT fanout on that socket.
203 c.replyOK(e.ID, true, reason)
204 if accessEvent(e) {
205 c.r.closeSubscriptionsLocked("blocked: relay access policy changed; subscribe again")
206 }
207 if reason == "" {
208 r.fanout(e)
209 }
210 return reason, nil
211}
212
213// BroadcastGenerated delivers an event already durably accepted by an
214// out-of-band producer.
215func (r *Router) BroadcastGenerated(e event.Event) {
216 r.mu.Lock()
217 defer r.mu.Unlock()
218 r.fanout(e)
219}
220
221// CommitGenerated runs an imported or server-generated event commit under the
222// same historical/live fence as client EVENT publication. The callback must
223// durably commit the event before returning; a returned error suppresses
224// fanout. This prevents a REQ snapshot from racing an out-of-band broadcast.
225func (r *Router) CommitGenerated(ctx context.Context, commit func(context.Context) (event.Event, error)) (event.Event, error) {
226 r.mu.Lock()
227 defer r.mu.Unlock()
228 e, err := commit(ctx)
229 if err == nil && e.ID != "" {
230 r.fanout(e)
231 }
232 return e, err
233}
234
235func (r *Router) Close(ctx context.Context) error {
236 r.closeOnce.Do(func() { close(r.closed) })
237 r.clientsMu.Lock()
238 cs := make([]*client, 0, len(r.clients))
239 for c := range r.clients {
240 cs = append(cs, c)
241 }
242 r.clientsMu.Unlock()
243 for _, c := range cs {
244 c.close(websocket.StatusGoingAway, "relay shutting down")
245 }
246 select {
247 case <-ctx.Done():
248 return ctx.Err()
249 case <-r.clientsDone():
250 return nil
251 }
252}
253
254// CloseSubscriptions invalidates all live subscriptions after a policy
255// revision. Existing sockets remain usable for AUTH and a subsequent REQ.
256func (r *Router) CloseSubscriptions(reason string) {
257 r.mu.Lock()
258 defer r.mu.Unlock()
259 r.closeSubscriptionsLocked(reason)
260}
261
262// ApplyACLChange serializes an access-control mutation with REQ snapshots,
263// NEG sessions, and live publication. A successful mutation invalidates both
264// subscription families before another transport operation can begin.
265func (r *Router) ApplyACLChange(ctx context.Context, change func(context.Context) (any, error), reason string) (any, error) {
266 r.mu.Lock()
267 defer r.mu.Unlock()
268 result, err := change(ctx)
269 if err == nil {
270 r.closeSubscriptionsLocked(reason)
271 }
272 return result, err
273}
274
275func (r *Router) closeSubscriptionsLocked(reason string) {
276 if reason == "" {
277 reason = "blocked: relay policy changed"
278 }
279 r.clientsMu.Lock()
280 cs := make([]*client, 0, len(r.clients))
281 for c := range r.clients {
282 cs = append(cs, c)
283 }
284 r.clientsMu.Unlock()
285 for _, c := range cs {
286 c.mu.Lock()
287 ids := make([]string, 0, len(c.subs))
288 for id := range c.subs {
289 ids = append(ids, id)
290 delete(c.subs, id)
291 }
292 syncIDs := make([]string, 0, len(c.syncs))
293 for id := range c.syncs {
294 syncIDs = append(syncIDs, id)
295 delete(c.syncs, id)
296 }
297 c.mu.Unlock()
298 for _, id := range ids {
299 if r.cfg.OnSubscription != nil {
300 r.cfg.OnSubscription(-1)
301 }
302 b, _ := json.Marshal([]any{"CLOSED", id, reason})
303 _ = c.enqueueBytes(b)
304 }
305 for _, id := range syncIDs {
306 b, _ := json.Marshal([]any{"NEG-CLOSE", id})
307 _ = c.enqueueBytes(b)
308 }
309 }
310 if p, ok := r.backend.(PolicyChanges); ok {
311 p.CloseSubscriptions(Session{RelayURL: r.cfg.RelayURL})
312 }
313}
314
315func (r *Router) clientsDone() <-chan struct{} {
316 done := make(chan struct{})
317 go func() { r.clientsWG.Wait(); close(done) }()
318 return done
319}
320
321type subscription struct{ filters []event.Filter }
322type outbound struct {
323 data []byte
324 size int
325}
326
327type client struct {
328 r *Router
329 conn *websocket.Conn
330 ctx context.Context
331 cancel context.CancelFunc
332 mu sync.Mutex
333 session Session
334 challenge string
335 subs map[string]subscription
336 syncs map[string]*syncprotocol.Session
337 queue []outbound
338 pending int
339 wake chan struct{}
340 done chan struct{}
341 once sync.Once
342 wg sync.WaitGroup
343}
344
345func newClient(r *Router, conn *websocket.Conn, req *http.Request) *client {
346 ctx, cancel := context.WithCancel(context.Background())
347 b := make([]byte, 16)
348 _, _ = rand.Read(b)
349 relayURL := r.cfg.RelayURL
350 if r.cfg.RequestRelayURL != nil {
351 if requested := strings.TrimSpace(r.cfg.RequestRelayURL(req)); requested != "" {
352 relayURL = requested
353 }
354 }
355 return &client{r: r, conn: conn, ctx: ctx, cancel: cancel, challenge: hex.EncodeToString(b), subs: make(map[string]subscription), syncs: make(map[string]*syncprotocol.Session), wake: make(chan struct{}, 1), done: make(chan struct{}), session: Session{RelayURL: relayURL, RemoteIP: remoteIP(req.RemoteAddr)}}
356}
357
358func (c *client) run() {
359 defer c.r.clientsWG.Done()
360 c.wg.Add(1)
361 go c.writer()
362 _ = c.enqueue(fmt.Sprintf(`["AUTH",%q]`, c.challenge))
363 c.readLoop()
364 c.close(websocket.StatusNormalClosure, "closed")
365 c.wg.Wait()
366}
367
368func (c *client) readLoop() {
369 defer c.conn.CloseNow()
370 readLimit := c.r.cfg.MaxMessageBytes
371 if readLimit == 0 {
372 readLimit = -1
373 }
374 c.conn.SetReadLimit(readLimit)
375 for {
376 // Read contexts must live for the connection. A per-message timeout
377 // measures time between application messages, so it closes a healthy
378 // idle session even when the websocket ping/pong control traffic is
379 // flowing. The writer owns protocol keepalive; cancellation closes a
380 // connection during shutdown.
381 typ, data, err := c.conn.Read(c.ctx)
382 if err != nil {
383 return
384 }
385 if typ != websocket.MessageText {
386 _ = c.enqueue(`["NOTICE","error: binary frames are not supported"]`)
387 continue
388 }
389 if c.r.cfg.MaxMessageBytes > 0 && int64(len(data)) > c.r.cfg.MaxMessageBytes {
390 _ = c.enqueue(`["NOTICE","error: message too large"]`)
391 continue
392 }
393 var raw []json.RawMessage
394 if err := json.Unmarshal(data, &raw); err != nil || len(raw) == 0 {
395 _ = c.enqueue(`["NOTICE","error: could not parse message"]`)
396 continue
397 }
398 var typName string
399 if json.Unmarshal(raw[0], &typName) != nil {
400 _ = c.enqueue(`["NOTICE","error: message type must be a string"]`)
401 continue
402 }
403 c.dispatch(typName, raw[1:])
404 }
405}
406
407func (c *client) dispatch(kind string, args []json.RawMessage) {
408 switch kind {
409 case "EVENT":
410 c.handleEvent(args)
411 case "REQ", "COUNT":
412 c.handleRequest(kind, args)
413 case "CLOSE":
414 c.handleClose(args)
415 case "AUTH":
416 c.handleAuth(args)
417 case "NEG-OPEN", "NEG-MSG", "NEG-CLOSE":
418 c.handleSync(kind, args)
419 default:
420 _ = c.enqueue(fmt.Sprintf(`["NOTICE","error: unknown message type %s"]`, kind))
421 }
422}
423
424func (c *client) handleSync(kind string, args []json.RawMessage) {
425 if len(args) < 1 {
426 _ = c.enqueue(fmt.Sprintf(`["NEG-ERR","",%q]`, "error: "+kind+" needs a subscription id"))
427 return
428 }
429 var id string
430 if json.Unmarshal(args[0], &id) != nil || id == "" || len(id) > 64 {
431 _ = c.enqueue(`["NEG-ERR","","error: bad subscription id"]`)
432 return
433 }
434 if kind == "NEG-CLOSE" {
435 c.mu.Lock()
436 delete(c.syncs, id)
437 c.mu.Unlock()
438 return
439 }
440 if kind == "NEG-OPEN" {
441 if len(args) < 3 {
442 c.syncError(id, "error: NEG-OPEN needs a filter and a message")
443 return
444 }
445 f, err := event.ParseFilter(args[1])
446 if err != nil {
447 c.syncError(id, "invalid: bad filter: "+err.Error())
448 return
449 }
450 var message string
451 if json.Unmarshal(args[2], &message) != nil {
452 c.syncError(id, "invalid: message must be hex")
453 return
454 }
455 if _, err := hex.DecodeString(message); err != nil {
456 c.syncError(id, "invalid: message must be hex")
457 return
458 }
459 b, ok := c.r.backend.(SyncBackend)
460 if !ok {
461 c.syncError(id, "unsupported: sync is switched off on this relay")
462 return
463 }
464 c.r.mu.RLock()
465 items, err := b.Sync(c.ctx, f, c.snapshot())
466 if err == nil {
467 var sess *syncprotocol.Session
468 sess, err = syncprotocol.NewSession(items, false)
469 if err == nil {
470 var result syncprotocol.Result
471 result, err = sess.Reconcile(message)
472 if err == nil {
473 if result.Response == "" {
474 _ = c.enqueue(fmt.Sprintf(`["NEG-CLOSE",%q]`, id))
475 } else {
476 c.mu.Lock()
477 c.syncs[id] = sess
478 c.mu.Unlock()
479 c.sendSyncMessage(id, result.Response)
480 }
481 }
482 }
483 }
484 c.r.mu.RUnlock()
485 if err != nil {
486 c.syncError(id, syncErrorReason(err))
487 }
488 return
489 }
490 if len(args) < 2 {
491 c.syncError(id, "error: NEG-MSG needs a message")
492 return
493 }
494 var message string
495 if json.Unmarshal(args[1], &message) != nil {
496 c.syncError(id, "invalid: message must be hex")
497 return
498 }
499 if _, err := hex.DecodeString(message); err != nil {
500 c.syncError(id, "invalid: message must be hex")
501 return
502 }
503 c.mu.Lock()
504 sess := c.syncs[id]
505 c.mu.Unlock()
506 if sess == nil {
507 c.syncError(id, "closed: no such sync, send NEG-OPEN first")
508 return
509 }
510 result, err := sess.Reconcile(message)
511 if err != nil {
512 c.mu.Lock()
513 delete(c.syncs, id)
514 c.mu.Unlock()
515 c.syncError(id, "invalid: "+err.Error())
516 return
517 }
518 if result.Response == "" {
519 c.mu.Lock()
520 delete(c.syncs, id)
521 c.mu.Unlock()
522 _ = c.enqueue(fmt.Sprintf(`["NEG-CLOSE",%q]`, id))
523 } else {
524 c.sendSyncMessage(id, result.Response)
525 }
526}
527
528func (c *client) sendSyncMessage(id, message string) {
529 b, _ := json.Marshal([]any{"NEG-MSG", id, message})
530 _ = c.enqueueBytes(b)
531}
532
533func (c *client) syncError(id, reason string) {
534 b, _ := json.Marshal([]any{"NEG-ERR", id, reason})
535 _ = c.enqueueBytes(b)
536}
537
538func (c *client) handleEvent(args []json.RawMessage) {
539 if len(args) < 1 {
540 _ = c.enqueue(`["NOTICE","error: EVENT needs an event"]`)
541 return
542 }
543 providedID := rawEventID(args[0])
544 e, err := event.Parse(args[0])
545 if err != nil {
546 c.replyOK(providedID, false, err.Error())
547 return
548 }
549 if e.Kind == event.KIND_AUTH {
550 c.replyOK(e.ID, false, "blocked: AUTH events use AUTH")
551 return
552 }
553 reason, err := c.r.publishClient(c, e, c.snapshot())
554 if err != nil {
555 c.replyOK(e.ID, false, err.Error())
556 return
557 }
558 _ = reason
559}
560
561func (c *client) handleRequest(kind string, args []json.RawMessage) {
562 if len(args) < 2 {
563 _ = c.enqueue(fmt.Sprintf(`["NOTICE",%q]`, "error: "+kind+" needs an id and at least one filter"))
564 return
565 }
566 var id string
567 if json.Unmarshal(args[0], &id) != nil || id == "" || len(id) > 64 {
568 _ = c.enqueue(`["NOTICE","error: bad subscription id"]`)
569 return
570 }
571 filters := make([]event.Filter, 0, len(args)-1)
572 for _, raw := range args[1:] {
573 f, err := event.ParseFilter(raw)
574 if err != nil {
575 c.closedSub(id, "invalid: bad filter: "+err.Error())
576 return
577 }
578 filters = append(filters, f)
579 }
580 if kind == "COUNT" {
581 c.handleCount(id, filters)
582 return
583 }
584 c.r.mu.Lock()
585 c.mu.Lock()
586 _, replaced := c.subs[id]
587 c.subs[id] = subscription{filters: filters}
588 if !replaced && c.r.cfg.OnSubscription != nil {
589 c.r.cfg.OnSubscription(1)
590 }
591 c.mu.Unlock()
592 var hints []string
593 var events []event.Event
594 var err error
595 if hinted, ok := c.r.backend.(QueryHints); ok {
596 events, hints, err = hinted.QueryHints(c.ctx, filters, c.snapshot())
597 } else {
598 events, err = c.r.backend.Query(c.ctx, filters, c.snapshot())
599 }
600 if err == nil {
601 for _, e := range events {
602 matched := false
603 for _, f := range filters {
604 if event.Matches(f, e) && c.canRead(e, &f) {
605 matched = true
606 break
607 }
608 }
609 if matched {
610 if !c.sendEvent(id, e) {
611 err = errors.New("slow consumer")
612 break
613 }
614 }
615 }
616 }
617 if err == nil {
618 if len(hints) > 0 {
619 _ = c.enqueue(fmt.Sprintf(`["EOSE",%q,%s]`, id, mustJSON(hints)))
620 } else {
621 _ = c.enqueue(fmt.Sprintf(`["EOSE",%q]`, id))
622 }
623 }
624 c.r.mu.Unlock()
625 if err != nil {
626 c.closedSub(id, queryErrorReason(err))
627 }
628}
629
630func (c *client) handleCount(id string, filters []event.Filter) {
631 c.r.mu.RLock()
632 value, err := c.r.backend.Count(c.ctx, filters, c.snapshot())
633 c.r.mu.RUnlock()
634 if err != nil {
635 c.closedSub(id, "error: count failed: "+err.Error())
636 return
637 }
638 _ = c.enqueue(fmt.Sprintf(`["COUNT",%q,%s]`, id, mustJSON(value)))
639}
640
641func (c *client) handleClose(args []json.RawMessage) {
642 if len(args) > 0 {
643 var id string
644 if json.Unmarshal(args[0], &id) == nil {
645 c.mu.Lock()
646 _, existed := c.subs[id]
647 delete(c.subs, id)
648 c.mu.Unlock()
649 if existed && c.r.cfg.OnSubscription != nil {
650 c.r.cfg.OnSubscription(-1)
651 }
652 }
653 }
654}
655
656func (c *client) handleAuth(args []json.RawMessage) {
657 if len(args) < 1 {
658 c.replyOK("", false, "invalid: AUTH needs an event")
659 return
660 }
661 providedID := rawEventID(args[0])
662 e, err := event.Parse(args[0])
663 if err != nil {
664 c.replyOK(providedID, false, err.Error())
665 return
666 }
667 if e.Kind != event.KIND_AUTH {
668 c.replyOK(e.ID, false, "invalid: auth event must be kind 22242")
669 return
670 }
671 if abs(time.Now().Unix()-e.CreatedAt) > 600 {
672 c.replyOK(e.ID, false, "invalid: auth event created_at is too far from the current time")
673 return
674 }
675 if event.Tag(e, "challenge") != c.challenge {
676 c.replyOK(e.ID, false, "invalid: auth challenge does not match")
677 return
678 }
679 if c.session.RelayURL != "" && normalizeAuthRelay(event.Tag(e, "relay")) != normalizeAuthRelay(c.session.RelayURL) {
680 c.replyOK(e.ID, false, "invalid: auth relay tag does not name this relay")
681 return
682 }
683 candidate := c.snapshot()
684 seen := false
685 for _, key := range candidate.PubKeys {
686 if key == e.PubKey {
687 seen = true
688 break
689 }
690 }
691 if !seen {
692 candidate.PubKeys = append(candidate.PubKeys, e.PubKey)
693 }
694 if c.r.cfg.OnAuthenticate != nil {
695 if err := c.r.cfg.OnAuthenticate(c.ctx, candidate); err != nil {
696 c.replyOK(e.ID, false, "error: authentication bookkeeping failed: "+err.Error())
697 return
698 }
699 }
700 if !seen {
701 c.mu.Lock()
702 c.session.PubKeys = append(c.session.PubKeys, e.PubKey)
703 c.mu.Unlock()
704 }
705 c.replyOK(e.ID, true, "")
706}
707
708func (r *Router) fanout(e event.Event) {
709 r.clientsMu.Lock()
710 cs := make([]*client, 0, len(r.clients))
711 for x := range r.clients {
712 cs = append(cs, x)
713 }
714 r.clientsMu.Unlock()
715 for _, x := range cs {
716 x.mu.Lock()
717 subs := make(map[string]subscription, len(x.subs))
718 for id, s := range x.subs {
719 subs[id] = s
720 }
721 x.mu.Unlock()
722 for id, s := range subs {
723 for _, f := range s.filters {
724 if event.Matches(f, e) && x.canRead(e, &f) {
725 if !x.sendEvent(id, e) {
726 x.close(websocket.StatusPolicyViolation, "slow consumer")
727 }
728 break
729 }
730 }
731 }
732 }
733 r.listenersMu.Lock()
734 ls := make([]*listener, 0, len(r.listeners))
735 for l := range r.listeners {
736 ls = append(ls, l)
737 }
738 r.listenersMu.Unlock()
739 for _, l := range ls {
740 l.fn(e)
741 }
742}
743
744func (c *client) canRead(e event.Event, f *event.Filter) bool {
745 if x, ok := c.r.backend.(FilterReader); ok {
746 return x.CanReadFilter(e, c.snapshot(), f)
747 }
748 if x, ok := c.r.backend.(Reader); ok {
749 return x.CanRead(e, c.snapshot())
750 }
751 return true
752}
753func (c *client) snapshot() Session {
754 c.mu.Lock()
755 defer c.mu.Unlock()
756 s := c.session
757 s.PubKeys = append([]string(nil), s.PubKeys...)
758 return s
759}
760func (c *client) sendEvent(id string, e event.Event) bool {
761 b, _ := json.Marshal([]any{"EVENT", id, e})
762 return c.enqueueBytes(b) == nil
763}
764func (c *client) replyOK(id string, ok bool, reason string) {
765 b, _ := json.Marshal([]any{"OK", id, ok, reason})
766 _ = c.enqueueBytes(b)
767}
768func (c *client) closedSub(id, reason string) {
769 c.mu.Lock()
770 delete(c.subs, id)
771 c.mu.Unlock()
772 b, _ := json.Marshal([]any{"CLOSED", id, reason})
773 _ = c.enqueueBytes(b)
774}
775func (c *client) enqueue(s string) error { return c.enqueueBytes([]byte(s)) }
776func (c *client) enqueueBytes(b []byte) error {
777 c.mu.Lock()
778 defer c.mu.Unlock()
779 if c.ctx.Err() != nil {
780 return c.ctx.Err()
781 }
782 if c.pending+len(b) > c.r.cfg.MaxPendingBytes {
783 return errors.New("slow consumer")
784 }
785 c.queue = append(c.queue, outbound{data: append([]byte(nil), b...), size: len(b)})
786 c.pending += len(b)
787 select {
788 case c.wake <- struct{}{}:
789 default:
790 }
791 return nil
792}
793
794func (c *client) writer() {
795 defer c.wg.Done()
796 ticker := time.NewTicker(c.r.cfg.PingInterval)
797 defer ticker.Stop()
798 for {
799 c.mu.Lock()
800 if len(c.queue) == 0 {
801 c.mu.Unlock()
802 select {
803 case <-c.wake:
804 case <-c.done:
805 return
806 case <-ticker.C:
807 ctx, cancel := context.WithTimeout(c.ctx, c.r.cfg.WriteLimit)
808 err := c.conn.Ping(ctx)
809 cancel()
810 if err != nil {
811 return
812 }
813 }
814 continue
815 }
816 item := c.queue[0]
817 c.queue = c.queue[1:]
818 c.pending -= item.size
819 c.mu.Unlock()
820 ctx, cancel := context.WithTimeout(c.ctx, c.r.cfg.WriteLimit)
821 err := c.conn.Write(ctx, websocket.MessageText, item.data)
822 cancel()
823 if err != nil {
824 return
825 }
826 }
827}
828func (c *client) close(code websocket.StatusCode, reason string) {
829 c.once.Do(func() {
830 close(c.done)
831 c.cancel()
832 _ = c.conn.Close(code, reason)
833 c.r.clientsMu.Lock()
834 delete(c.r.clients, c)
835 c.r.clientsMu.Unlock()
836 if c.r.cfg.OnConnection != nil {
837 c.r.cfg.OnConnection(-1)
838 }
839 c.mu.Lock()
840 subscriptions := len(c.subs)
841 c.subs = make(map[string]subscription)
842 c.mu.Unlock()
843 if c.r.cfg.OnSubscription != nil && subscriptions > 0 {
844 c.r.cfg.OnSubscription(-subscriptions)
845 }
846 })
847}
848func mustJSON(v any) string {
849 b, _ := json.Marshal(v)
850 if len(b) == 0 {
851 return "null"
852 }
853 return string(b)
854}
855func abs(v int64) int64 {
856 if v < 0 {
857 return -v
858 }
859 return v
860}
861func normalizeRelay(v string) string {
862 u, err := url.Parse(strings.TrimSpace(v))
863 if err != nil || u.Hostname() == "" {
864 return ""
865 }
866 return strings.ToLower(u.Hostname())
867}
868
869func normalizeAuthRelay(v string) string {
870 u, err := url.Parse(strings.TrimSpace(v))
871 if err != nil || u.Scheme == "" || u.Host == "" || u.User != nil || u.Fragment != "" || u.Opaque != "" {
872 return ""
873 }
874 host := strings.ToLower(u.Hostname())
875 if port := u.Port(); port != "" {
876 host += ":" + port
877 }
878 path := strings.TrimRight(u.EscapedPath(), "/")
879 if path == "" {
880 path = "/"
881 }
882 return strings.ToLower(u.Scheme) + "://" + host + path
883}
884
885func remoteIP(address string) string {
886 if host, _, err := net.SplitHostPort(strings.TrimSpace(address)); err == nil {
887 return host
888 }
889 if ip := net.ParseIP(strings.TrimSpace(address)); ip != nil {
890 return ip.String()
891 }
892 return strings.TrimSpace(address)
893}
894
895func rawEventID(raw json.RawMessage) string {
896 var probe struct {
897 ID string `json:"id"`
898 }
899 if json.Unmarshal(raw, &probe) != nil {
900 return ""
901 }
902 return probe.ID
903}
904
905func queryErrorReason(err error) string {
906 if err == nil {
907 return ""
908 }
909 if knownProtocolReason(err.Error()) {
910 return err.Error()
911 }
912 return "error: query failed: " + err.Error()
913}
914
915func syncErrorReason(err error) string {
916 if err == nil {
917 return ""
918 }
919 if knownProtocolReason(err.Error()) {
920 return err.Error()
921 }
922 return "error: sync failed: " + err.Error()
923}
924
925func knownProtocolReason(reason string) bool {
926 for _, prefix := range []string{"auth-required:", "restricted:", "blocked:", "invalid:", "unsupported:"} {
927 if strings.HasPrefix(reason, prefix) {
928 return true
929 }
930 }
931 return false
932}
933
934func accessEvent(e event.Event) bool {
935 if e.Kind == event.KIND_NIP43_JOIN && event.Tag(e, "claim") == "" {
936 // An access request waits for review and changes nobody's access.
937 return false
938 }
939 switch e.Kind {
940 case event.KIND_REPORT, event.KIND_VANISH, event.KIND_JOIN, event.KIND_LEAVE,
941 event.KIND_NIP43_JOIN, event.KIND_NIP43_LEAVE, event.KIND_PUT_USER,
942 event.KIND_REMOVE_USER, event.KIND_DELETE_EVENT, event.KIND_DELETE_GROUP:
943 return true
944 default:
945 return false
946 }
947}
948
949func duplicateReason(reason string) bool {
950 return strings.HasPrefix(reason, "duplicate:")
951}

About

ownerc01503e8035f
visibilitypublic
headd730679639bf

Maintainers

  • c01503e8035f owner
  • ba1b5beed5b5 agent hermes

Clone

  • https://012.run/npub1cq2s86qrtadf36eqq3epm38wmhdynrrcwxf84eldk3f94dkv7ttsrulrz5/tinyrelay.git

Refs