client.mx raw
1 package ws
2
3 import (
4 "fmt"
5 "sync"
6
7 "git.smesh.lol/nostr/pkg/envelope"
8 "git.smesh.lol/nostr/pkg/event"
9 "git.smesh.lol/nostr/pkg/filter"
10 "git.smesh.lol/nostr/pkg/text"
11 )
12
13 // Client is a Nostr relay client.
14 type Client struct {
15 ws *Conn
16 url string
17 subs map[string]*Sub
18 mu sync.Mutex
19 idN int64
20 done chan struct{}
21 OKs chan *envelope.OK // buffered channel for OK responses
22 Err error // set when the read loop exits
23 }
24
25 // Sub is an active subscription.
26 type Sub struct {
27 ID string
28 Events chan *event.E
29 EOSE chan struct{}
30 eosed bool
31 }
32
33 // Connect dials a Nostr relay. Call RunReadLoop() to start processing
34 // incoming messages (blocks until the connection closes).
35 func Connect(rawURL string) (cl *Client, derr error) {
36 ws, err := Dial(rawURL)
37 if err != nil {
38 return nil, err
39 }
40 c := &Client{
41 ws: ws,
42 url: rawURL,
43 subs: map[string]*Sub{},
44 done: chan struct{}{},
45 OKs: chan *envelope.OK{64},
46 }
47 return c, nil
48 }
49
50 // RunReadLoop reads messages from the relay and dispatches them to
51 // subscriptions. Blocks until the connection closes.
52 func (c *Client) RunReadLoop() {
53 defer close(c.done)
54 var op byte
55 var payload []byte
56 var err error
57 for {
58 op, payload, err = c.ws.ReadMessage()
59 if err != nil {
60 c.Err = err
61 return
62 }
63 if op == OpClose {
64 return
65 }
66 if op != OpText {
67 continue
68 }
69 c.dispatch(payload)
70 }
71 }
72
73 // Subscribe sends a REQ and returns a subscription that delivers events.
74 func (c *Client) Subscribe(filters ...*filter.F) (sub2 *Sub, serr error) {
75 c.mu.Lock()
76 c.idN++
77 id := string([]byte(nil) | fmt.Sprintf("s%d", c.idN))
78 sub := &Sub{
79 ID: id,
80 Events: chan *event.E{256},
81 EOSE: chan struct{}{1},
82 }
83 c.subs[id] = sub
84 c.mu.Unlock()
85
86 req := &envelope.Req{
87 Subscription: []byte(id),
88 Filters: filter.S{F: filters},
89 }
90 if err := c.ws.WriteText(req.Marshal(nil)); err != nil {
91 c.mu.Lock()
92 delete(c.subs, id)
93 c.mu.Unlock()
94 return nil, err
95 }
96 return sub, nil
97 }
98
99 // Publish sends an event to the relay.
100 func (c *Client) Publish(ev *event.E) (err error) {
101 es := &envelope.EventSubmission{E: ev}
102 return c.ws.WriteText(es.Marshal(nil))
103 }
104
105 // Unsubscribe sends CLOSE and removes the subscription.
106 func (c *Client) Unsubscribe(sub *Sub) (err error) {
107 cl := &envelope.Close{ID: []byte(sub.ID)}
108 if err = c.ws.WriteText(cl.Marshal(nil)); err != nil {
109 return err
110 }
111 c.mu.Lock()
112 delete(c.subs, sub.ID)
113 c.mu.Unlock()
114 return nil
115 }
116
117 // Close shuts down the client.
118 func (c *Client) Close() (err error) {
119 c.mu.Lock()
120 for _, sub := range c.subs {
121 close(sub.Events)
122 }
123 c.subs = nil
124 c.mu.Unlock()
125 return c.ws.Close()
126 }
127
128 // Done returns a channel closed when the read loop exits.
129 func (c *Client) Done() (v <-chan struct{}) { return c.done }
130
131 func (c *Client) dispatch(msg []byte) {
132 label, rem, err := envelope.Identify(msg)
133 if err != nil {
134 return
135 }
136 switch label {
137 case envelope.EventLabel:
138 c.handleEvent(rem)
139 case envelope.EOSELabel:
140 c.handleEOSE(rem)
141 case envelope.OKLabel:
142 c.handleOK(rem)
143 case envelope.ClosedLabel:
144 c.handleClosed(rem)
145 }
146 }
147
148 func (c *Client) handleEvent(rem []byte) {
149 subID, rem, err := text.UnmarshalQuoted(rem)
150 if err != nil {
151 return
152 }
153 // skip comma
154 for len(rem) > 0 && (rem[0] == ',' || rem[0] == ' ') {
155 rem = rem[1:]
156 }
157 ev := event.New()
158 if _, err = ev.Unmarshal(rem); err != nil {
159 return
160 }
161 c.mu.Lock()
162 sub := c.subs[string(subID)]
163 c.mu.Unlock()
164 if sub != nil {
165 select {
166 case sub.Events <- ev:
167 default:
168 }
169 }
170 }
171
172 func (c *Client) handleEOSE(rem []byte) {
173 subID, _, err := text.UnmarshalQuoted(rem)
174 if err != nil {
175 return
176 }
177 c.mu.Lock()
178 sub := c.subs[string(subID)]
179 c.mu.Unlock()
180 if sub != nil && !sub.eosed {
181 sub.eosed = true
182 select {
183 case sub.EOSE <- struct{}{}:
184 default:
185 }
186 }
187 }
188
189 func (c *Client) handleOK(rem []byte) {
190 var ok envelope.OK
191 if _, err := ok.Unmarshal(rem); err != nil {
192 return
193 }
194 select {
195 case c.OKs <- &ok:
196 default:
197 }
198 }
199
200 func (c *Client) handleClosed(rem []byte) {
201 subID, _, err := text.UnmarshalQuoted(rem)
202 if err != nil {
203 return
204 }
205 c.mu.Lock()
206 sub := c.subs[string(subID)]
207 if sub != nil {
208 delete(c.subs, string(subID))
209 }
210 c.mu.Unlock()
211 if sub != nil {
212 close(sub.Events)
213 }
214 }
215