client.go raw
1 // Package relay is a small NIP-01 client for one relay connection.
2 //
3 // It owns the text frames, the subscription bookkeeping and the event
4 // signature check. Reconnection is the caller's policy: a Client is one
5 // connection, and when Run returns every subscription has ended.
6 package relay
7
8 import (
9 "context"
10 "errors"
11 "fmt"
12 "sync"
13 "time"
14
15 "git.smesh.lol/nostrino/pkg/nostr"
16 "git.smesh.lol/nostrino/pkg/ws"
17 )
18
19 // Transport is one WebSocket connection carrying text messages. The ws
20 // package implements it for native builds; a syscall/js adapter can implement
21 // it for wasm.
22 type Transport interface {
23 WriteText([]byte) error
24 ReadMessage() ([]byte, error)
25 Close() error
26 }
27
28 // DialTransport opens a ws:// or wss:// transport.
29 func DialTransport(rawURL string) (Transport, error) {
30 return ws.Dial(rawURL)
31 }
32
33 // Options bound a relay connection.
34 type Options struct {
35 // EventBuffer is the per-subscription queue length. Zero means 64.
36 EventBuffer int
37 // OnDrop is called when a subscription queue is full. It must not block.
38 OnDrop func(subID string, dropped int)
39 // RelayURL is the URL signed into the NIP-42 auth event. Empty means the
40 // relay URL is not available and AUTH challenges cannot be answered
41 // correctly; a relay that requires auth will reject the connection.
42 RelayURL string
43 }
44
45 // Client is one authenticated relay connection.
46 type Client struct {
47 t Transport
48 opts Options
49
50 key [32]byte
51 pub [32]byte
52
53 writeMu sync.Mutex
54 mu sync.Mutex
55 subs map[string]*Subscription
56 oks map[string]chan string
57 nextID uint64
58 closed bool
59 err error
60 done chan struct{}
61
62 onClose func(error)
63
64 relayURL string
65 authState int // 0 unseen, 1 pending, 2 authenticated, 3 failed
66 authID string
67 authReady chan struct{}
68 authDone bool
69 authErr error
70 }
71
72 // New starts a relay client and its read loop. The caller must not use t
73 // afterwards. key signs the client's own events.
74 func New(t Transport, key [32]byte, opts Options) (*Client, error) {
75 pub, ok := nostrPubKey(key)
76 if !ok {
77 return nil, errors.New("relay: invalid secret key")
78 }
79 if opts.EventBuffer <= 0 {
80 opts.EventBuffer = 64
81 }
82 c := &Client{
83 t: t,
84 opts: opts,
85 key: key,
86 pub: pub,
87 subs: make(map[string]*Subscription),
88 oks: make(map[string]chan string),
89 done: make(chan struct{}),
90 relayURL: opts.RelayURL,
91 }
92 go c.readLoop()
93 return c, nil
94 }
95
96 func nostrPubKey(key [32]byte) ([32]byte, bool) {
97 var zero [32]byte
98 // The nostr package validates and derives the key through its own helper.
99 h, err := nostr.PubKeyHex(key)
100 if err != nil || h == "" {
101 return zero, false
102 }
103 var pub [32]byte
104 if err := nostr.Decode32(h, pub[:]); err != nil {
105 return zero, false
106 }
107 return pub, true
108 }
109
110 // PubKey is the client's public key.
111 func (c *Client) PubKey() [32]byte { return c.pub }
112
113 // PubKeyHex is the client's public key as lowercase hex.
114 func (c *Client) PubKeyHex() string { return nostr.Hex(c.pub[:]) }
115
116 // Done is closed when the read loop ends.
117 func (c *Client) Done() <-chan struct{} { return c.done }
118
119 // Err returns the read-loop error after Done is closed.
120 func (c *Client) Err() error {
121 c.mu.Lock()
122 defer c.mu.Unlock()
123 return c.err
124 }
125
126 // OnClose installs a callback that runs once when the connection ends.
127 func (c *Client) OnClose(fn func(error)) {
128 c.mu.Lock()
129 c.onClose = fn
130 c.mu.Unlock()
131 }
132
133 // Subscription is one REQ and its delivered events.
134 type Subscription struct {
135 ID string
136 Filter nostr.Filter
137
138 Events chan *nostr.Event
139 EOSE chan struct{}
140
141 once sync.Once
142 eoseOnce sync.Once
143 done chan struct{}
144
145 client *Client
146 }
147
148 // Done is closed when the subscription is closed or the connection ends.
149 func (s *Subscription) Done() <-chan struct{} { return s.done }
150
151 // Close sends CLOSE and removes the subscription.
152 func (s *Subscription) Close() error { return s.client.unsubscribe(s.ID) }
153
154 // Err returns the connection error once Done is closed.
155 func (s *Subscription) Err() error { return s.client.Err() }
156
157 func (s *Subscription) finish() {
158 s.once.Do(func() {
159 close(s.done)
160 s.signalEOSE()
161 })
162 }
163
164 func (s *Subscription) signalEOSE() {
165 s.eoseOnce.Do(func() { close(s.EOSE) })
166 }
167
168 // Subscribe sends a REQ and returns the subscription. It does not wait for
169 // EOSE; call WaitEOSE when the subscription must be live before the peer is
170 // addressed.
171 // awaitAuth blocks while a NIP-42 authentication is in flight. It returns nil
172 // when no auth has been seen or auth succeeded.
173 func (c *Client) awaitAuth(ctx context.Context) error {
174 c.mu.Lock()
175 state, ch, err := c.authState, c.authReady, c.authErr
176 c.mu.Unlock()
177 if state != 1 {
178 return err
179 }
180 if ch == nil {
181 return nil
182 }
183 select {
184 case <-ch:
185 c.mu.Lock()
186 err := c.authErr
187 c.mu.Unlock()
188 return err
189 case <-ctx.Done():
190 return ctx.Err()
191 case <-c.done:
192 return c.Err()
193 }
194 }
195
196 func (c *Client) sendAuth(challenge string) {
197 if c.relayURL == "" {
198 return
199 }
200 ev := AuthEvent(c.relayURL, challenge, c.key)
201 if ev == nil {
202 return
203 }
204 c.mu.Lock()
205 c.authState = 1
206 c.authID = nostr.Hex(ev.ID[:])
207 c.authReady = make(chan struct{})
208 c.authDone = false
209 c.authErr = nil
210 c.mu.Unlock()
211
212 msg := append([]byte(`["AUTH",`), ev.AppendJSON(nil)...)
213 msg = append(msg, ']')
214 if err := c.write(msg); err != nil {
215 c.authFinished(err)
216 }
217 }
218
219 func (c *Client) authFinished(err error) {
220 c.mu.Lock()
221 if c.authDone {
222 c.mu.Unlock()
223 return
224 }
225 c.authDone = true
226 c.authErr = err
227 if err != nil {
228 c.authState = 3
229 } else {
230 c.authState = 2
231 }
232 ch := c.authReady
233 c.mu.Unlock()
234 if ch != nil {
235 close(ch)
236 }
237 }
238
239 // AuthEvent builds and signs a NIP-42 authentication event.
240 func AuthEvent(relayURL, challenge string, key [32]byte) *nostr.Event {
241 ev := &nostr.Event{
242 Kind: 22242,
243 CreatedAt: time.Now().Unix(),
244 Tags: nostr.Tags{
245 {"relay", relayURL},
246 {"challenge", challenge},
247 },
248 }
249 if err := ev.Sign(key); err != nil {
250 return nil
251 }
252 return ev
253 }
254
255 func (c *Client) Subscribe(f nostr.Filter) (*Subscription, error) {
256 c.mu.Lock()
257 if c.closed {
258 err := c.err
259 c.mu.Unlock()
260 if err == nil {
261 err = errors.New("relay: closed")
262 }
263 return nil, err
264 }
265 c.nextID++
266 id := fmt.Sprintf("n%x", c.nextID)
267 s := &Subscription{
268 ID: id,
269 Filter: f,
270 Events: make(chan *nostr.Event, c.opts.EventBuffer),
271 EOSE: make(chan struct{}),
272 done: make(chan struct{}),
273 client: c,
274 }
275 c.subs[id] = s
276 c.mu.Unlock()
277
278 msg := []byte(`["REQ","` + id + `",`)
279 msg = f.AppendJSON(msg)
280 msg = append(msg, ']')
281 if err := c.write(msg); err != nil {
282 c.mu.Lock()
283 delete(c.subs, id)
284 c.mu.Unlock()
285 return nil, err
286 }
287 return s, nil
288 }
289
290 // WaitEOSE waits until the relay has replayed the stored matches, or ctx ends.
291 func (c *Client) WaitEOSE(ctx context.Context, s *Subscription) error {
292 select {
293 case <-s.EOSE:
294 return nil
295 case <-s.done:
296 if err := c.Err(); err != nil {
297 return err
298 }
299 return errors.New("relay: subscription closed")
300 case <-ctx.Done():
301 return ctx.Err()
302 }
303 }
304
305 func (c *Client) unsubscribe(id string) error {
306 c.mu.Lock()
307 s, ok := c.subs[id]
308 if ok {
309 delete(c.subs, id)
310 }
311 c.mu.Unlock()
312 if !ok {
313 return nil
314 }
315 err := c.write([]byte(`["CLOSE","` + id + `"]`))
316 s.finish()
317 return err
318 }
319
320 // Send signs and writes an event without waiting for the relay's OK.
321 // It is the data-plane path: the tunnel's own acknowledgements decide
322 // delivery, and waiting for a per-frame OK would add a round trip per frame.
323 func (c *Client) Send(ev *nostr.Event) error {
324 actx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
325 err := c.awaitAuth(actx)
326 cancel()
327 if err != nil {
328 return err
329 }
330 if err := ev.Sign(c.key); err != nil {
331 return err
332 }
333 msg := append([]byte(`["EVENT",`), ev.AppendJSON(nil)...)
334 msg = append(msg, ']')
335 return c.write(msg)
336 }
337
338 // Publish signs and writes an event and waits for the relay's OK.
339 func (c *Client) Publish(ctx context.Context, ev *nostr.Event) error {
340 if err := c.awaitAuth(ctx); err != nil {
341 return err
342 }
343 if err := ev.Sign(c.key); err != nil {
344 return err
345 }
346 ch := make(chan string, 1)
347 id := nostr.Hex(ev.ID[:])
348 c.mu.Lock()
349 c.oks[id] = ch
350 c.mu.Unlock()
351 defer func() {
352 c.mu.Lock()
353 delete(c.oks, id)
354 c.mu.Unlock()
355 }()
356
357 msg := append([]byte(`["EVENT",`), ev.AppendJSON(nil)...)
358 msg = append(msg, ']')
359 if err := c.write(msg); err != nil {
360 return err
361 }
362 select {
363 case reason := <-ch:
364 if reason != "" {
365 return errors.New("relay: rejected: " + reason)
366 }
367 return nil
368 case <-c.done:
369 if err := c.Err(); err != nil {
370 return err
371 }
372 return errors.New("relay: closed")
373 case <-ctx.Done():
374 return ctx.Err()
375 }
376 }
377
378 // Close closes the transport and ends the read loop.
379 func (c *Client) Close() error {
380 c.mu.Lock()
381 if c.closed {
382 c.mu.Unlock()
383 return nil
384 }
385 c.closed = true
386 c.mu.Unlock()
387 err := c.t.Close()
388 // Release waiters at once rather than waiting for the read loop to observe
389 // the closed socket. fail is idempotent, so the read loop's own report is
390 // harmless.
391 c.fail(errors.New("relay: closed"))
392 return err
393 }
394
395 func (c *Client) write(msg []byte) error {
396 c.writeMu.Lock()
397 defer c.writeMu.Unlock()
398 if err := c.t.WriteText(msg); err != nil {
399 c.fail(err)
400 return err
401 }
402 return nil
403 }
404
405 func (c *Client) readLoop() {
406 for {
407 raw, err := c.t.ReadMessage()
408 if err != nil {
409 c.fail(err)
410 return
411 }
412 if err := c.handle(raw); err != nil {
413 c.fail(err)
414 return
415 }
416 }
417 }
418
419 func (c *Client) fail(err error) {
420 c.mu.Lock()
421 if c.closed {
422 c.mu.Unlock()
423 return
424 }
425 c.closed = true
426 c.err = err
427 subs := make([]*Subscription, 0, len(c.subs))
428 for _, s := range c.subs {
429 subs = append(subs, s)
430 }
431 c.subs = map[string]*Subscription{}
432 oks := c.oks
433 c.oks = map[string]chan string{}
434 authCh := c.authReady
435 authDone := c.authDone
436 if !authDone && authCh != nil {
437 c.authDone = true
438 c.authState = 3
439 c.authErr = err
440 }
441 fn := c.onClose
442 c.mu.Unlock()
443 if !authDone && authCh != nil {
444 close(authCh)
445 }
446
447 close(c.done)
448 for _, s := range subs {
449 s.finish()
450 }
451 for id, ch := range oks {
452 select {
453 case ch <- "connection closed":
454 default:
455 _ = id
456 }
457 }
458 c.t.Close()
459 if fn != nil {
460 fn(err)
461 }
462 }
463
464 func (c *Client) handle(raw []byte) error {
465 parts, err := nostr.SplitArray(raw)
466 if err != nil {
467 return fmt.Errorf("relay: bad message: %w", err)
468 }
469 if len(parts) == 0 {
470 return errors.New("relay: empty message")
471 }
472 typ, err := nostr.ParseJSONString(parts[0])
473 if err != nil {
474 return fmt.Errorf("relay: bad message type: %w", err)
475 }
476 switch typ {
477 case "EVENT":
478 if len(parts) != 3 {
479 return errors.New("relay: malformed EVENT")
480 }
481 subID, err := nostr.ParseJSONString(parts[1])
482 if err != nil {
483 return fmt.Errorf("relay: malformed EVENT subscription: %w", err)
484 }
485 ev, err := nostr.ParseEvent(parts[2])
486 if err != nil {
487 return fmt.Errorf("relay: malformed event: %w", err)
488 }
489 if err := ev.Verify(); err != nil {
490 return fmt.Errorf("relay: event verification: %w", err)
491 }
492 c.mu.Lock()
493 s := c.subs[subID]
494 c.mu.Unlock()
495 if s == nil {
496 return nil
497 }
498 if !s.Filter.Matches(ev) {
499 return nil
500 }
501 select {
502 case s.Events <- ev:
503 default:
504 if c.opts.OnDrop != nil {
505 c.opts.OnDrop(subID, 1)
506 }
507 }
508 return nil
509 case "EOSE":
510 if len(parts) < 2 {
511 return errors.New("relay: malformed EOSE")
512 }
513 subID, err := nostr.ParseJSONString(parts[1])
514 if err != nil {
515 return err
516 }
517 c.mu.Lock()
518 s := c.subs[subID]
519 c.mu.Unlock()
520 if s != nil {
521 s.signalEOSE()
522 }
523 return nil
524 case "OK":
525 if len(parts) < 3 {
526 return errors.New("relay: malformed OK")
527 }
528 id, err := nostr.ParseJSONString(parts[1])
529 if err != nil {
530 return err
531 }
532 accepted := string(parts[2]) == "true"
533 reason := ""
534 if !accepted {
535 reason = "rejected"
536 if len(parts) >= 4 {
537 if m, err := nostr.ParseJSONString(parts[3]); err == nil && m != "" {
538 reason = m
539 }
540 }
541 }
542 c.mu.Lock()
543 ch := c.oks[id]
544 authID := c.authID
545 c.mu.Unlock()
546 if id == authID && authID != "" {
547 if reason == "" {
548 c.authFinished(nil)
549 } else {
550 c.authFinished(errors.New("relay: authentication rejected: " + reason))
551 }
552 }
553 if ch != nil {
554 select {
555 case ch <- reason:
556 default:
557 }
558 }
559 return nil
560 case "AUTH":
561 if len(parts) < 2 {
562 return errors.New("relay: malformed AUTH")
563 }
564 challenge, err := nostr.ParseJSONString(parts[1])
565 if err != nil {
566 return err
567 }
568 c.mu.Lock()
569 state := c.authState
570 c.mu.Unlock()
571 if state != 1 && state != 2 {
572 c.sendAuth(challenge)
573 }
574 return nil
575 case "NOTICE", "CLOSED":
576 // NOTICE is advisory; CLOSED ends one subscription; AUTH is not used
577 // by the public data plane. Torly closes only invalid subscriptions,
578 // which the tunnel treats as a connection failure through its
579 // acknowledgements anyway.
580 if typ == "CLOSED" && len(parts) >= 2 {
581 if subID, err := nostr.ParseJSONString(parts[1]); err == nil {
582 c.mu.Lock()
583 s := c.subs[subID]
584 delete(c.subs, subID)
585 c.mu.Unlock()
586 if s != nil {
587 s.finish()
588 }
589 }
590 }
591 return nil
592 default:
593 // Unknown message types are ignored, as NIP-01 requires.
594 return nil
595 }
596 }
597