1 package relay
2 3 import (
4 "runtime"
5 "git.smesh.lol/musiquay/web/common/jsbridge/ws"
6 "git.smesh.lol/nostr/pkg/core"
7 )
8 9 // State constants for connection readiness.
10 const (
11 StateConnecting = 0
12 StateOpen = 1
13 StateClosed = 2
14 )
15 16 // Conn is a single relay WebSocket connection.
17 type Conn struct {
18 URL string
19 wsConn ws.Conn
20 state int32
21 subs map[string]*Sub
22 23 // Callbacks set before Dial returns.
24 onReady func(bool)
25 26 onEvent func(string, *nostr.Event)
27 onEOSE func(string)
28 onOK func(string, bool, string)
29 onAuth func(string)
30 31 closing bool // true when Close() was called intentionally
32 everOpened bool // false until the first successful open
33 pendingSend []string // REQ/CLOSE queued when the socket was not writable
34 failCount int32 // consecutive connection failures for backoff
35 36 // ScheduleReconnect is set by the consumer to provide a delayed callback.
37 // Receives the desired delay in milliseconds so the consumer can apply it.
38 ScheduleReconnect func(delayMs int32, fn func())
39 }
40 41 // Dial opens a connection to a relay.
42 // Call OnReady to receive the open/fail notification, then Start to begin processing.
43 func Dial(url string) (c *Conn) {
44 // The connection outlives this frame: the caller keeps it for the life of
45 // the connection, and the ws callbacks registered in dial() fire long
46 // after Dial returns. A frame-arena Conn is copied into the caller's
47 // arena on return, which leaves both the callbacks and the address-keyed
48 // sovereign arena that holds them bound to the old address - the caller's
49 // Subscribe() writes then land in a different object than the one the open
50 // callback flushes.
51 prev := runtime.CurrentArena()
52 runtime.SovereignSetArena(runtime.RootArena())
53 c = &Conn{
54 URL: url,
55 state: StateConnecting,
56 subs: map[string]*Sub{},
57 }
58 59 c.dial()
60 runtime.SovereignRestoreArena(prev)
61 return c
62 }
63 64 func (c *Conn) dial() {
65 c.wsConn = ws.Dial(
66 c.URL,
67 func(connID int32, data string) {
68 c.handleMessage(data)
69 },
70 func(connID int32) {
71 c.state = StateOpen
72 c.failCount = 0 // reset backoff on successful connect
73 first := !c.everOpened
74 c.everOpened = true
75 if c.onReady != nil {
76 c.onReady(true)
77 c.onReady = nil
78 }
79 // Only a RE-open has to re-establish subscriptions. On the first
80 // open every REQ is already in hand: Subscribe sent it, or queued it
81 // in pendingSend when the socket was not writable. Flushing here as
82 // well sent each REQ a second time - and that second send went out
83 // as a zero-filled buffer of the same length, which the relay could
84 // not parse and which stopped it answering that connection, so the
85 // later subscriptions on it (the feed) never got their stored event.
86 c.flushPendingSend()
87 if !first {
88 c.flushSubs()
89 }
90 },
91 func(connID int32, code int32, reason string) {
92 c.state = StateClosed
93 if c.onReady != nil {
94 c.onReady(false)
95 c.onReady = nil
96 }
97 c.maybeReconnect()
98 },
99 func(connID int32) {
100 c.state = StateClosed
101 if c.onReady != nil {
102 c.onReady(false)
103 c.onReady = nil
104 }
105 c.maybeReconnect()
106 },
107 )
108 }
109 110 func (c *Conn) maybeReconnect() {
111 if c.closing || len(c.subs) == 0 || c.ScheduleReconnect == nil {
112 return
113 }
114 c.failCount++
115 // Exponential backoff: 5s → 30s → 5min → 30min cap.
116 // Dead relays stop spamming logs after a few attempts.
117 var delayMs int32
118 switch {
119 case c.failCount <= 2:
120 delayMs = 5000
121 case c.failCount <= 5:
122 delayMs = 30000
123 case c.failCount <= 8:
124 delayMs = 300000 // 5 min
125 default:
126 delayMs = 1800000 // 30 min
127 }
128 c.state = StateConnecting // prevent pool from creating a duplicate
129 c.ScheduleReconnect(delayMs, func() {
130 if c.closing {
131 return
132 }
133 c.dial()
134 })
135 }
136 137 // OnReady sets a callback that fires once when the connection opens (true) or fails (false).
138 func (c *Conn) OnReady(fn func(bool)) {
139 if c.state == StateOpen {
140 fn(true)
141 return
142 }
143 if c.state == StateClosed {
144 fn(false)
145 return
146 }
147 c.onReady = fn
148 }
149 150 // IsOpen returns whether the connection is open.
151 func (c *Conn) IsOpen() (ok bool) {
152 return c.state == StateOpen
153 }
154 155 func (c *Conn) handleMessage(msg string) {
156 label, subID, payload := nostr.ParseRelayMessage(msg)
157 158 switch label {
159 case "EVENT":
160 ev := nostr.ParseEvent(payload)
161 if ev == nil {
162 return
163 }
164 if sub, ok2 := c.subs[subID]; ok2 {
165 if sub.OnEvent != nil {
166 sub.OnEvent(ev)
167 }
168 }
169 if c.onEvent != nil {
170 c.onEvent(subID, ev)
171 }
172 173 case "EOSE":
174 if sub, ok2 := c.subs[subID]; ok2 {
175 sub.gotEOSE = true
176 if sub.OnEOSE != nil {
177 sub.OnEOSE()
178 }
179 }
180 if c.onEOSE != nil {
181 c.onEOSE(subID)
182 }
183 184 case "OK":
185 ok := len(payload) > 0 && payload[0] == 't'
186 reason := ""
187 idx := indexOf(payload, ':')
188 if idx >= 0 && idx+1 < len(payload) {
189 reason = payload[idx+1:]
190 }
191 if c.onOK != nil {
192 c.onOK(subID, ok, reason)
193 }
194 195 case "AUTH":
196 if c.onAuth != nil {
197 c.onAuth(payload)
198 }
199 200 case "NOTICE":
201 _ = payload
202 }
203 }
204 205 // Subscribe sends a REQ and tracks the subscription.
206 // If the connection is still opening, the REQ is deferred to flushSubs().
207 func (c *Conn) Subscribe(id string, filters []*nostr.Filter) (s *Sub) {
208 // The subscription outlives this frame: the Conn keeps it in c.subs and the
209 // ws callbacks read it long after Subscribe returns. Build it, and the id it
210 // is keyed by, in the root arena the Conn itself lives in. filters are the
211 // caller's: they must already live somewhere that outlives this call.
212 prev := runtime.CurrentArena()
213 runtime.SovereignSetArena(runtime.RootArena())
214 idCopy := []byte{:len(id)}
215 copy(idCopy, id)
216 sub := &Sub{
217 ID: string(idCopy),
218 Filters: filters,
219 conn: c,
220 }
221 c.subs[sub.ID] = sub
222 runtime.SovereignRestoreArena(prev)
223 224 msg := "[\"REQ\",\"" | sub.ID | "\""
225 for _, f := range filters {
226 msg |= "," | f.Serialize()
227 }
228 msg |= "]"
229 c.sendOrQueue(msg)
230 231 return sub
232 }
233 234 // Publish sends an EVENT message. Queues if the connection is still opening.
235 func (c *Conn) Publish(ev *nostr.Event) {
236 msg := "[\"EVENT\"," | eventJSON(ev) | "]"
237 if c.state != StateOpen {
238 c.queueSend(msg)
239 return
240 }
241 c.sendOrQueue(msg)
242 }
243 244 // CloseSubscription sends a CLOSE message.
245 func (c *Conn) CloseSubscription(id string) {
246 delete(c.subs, id)
247 msg := "[\"CLOSE\",\"" | id | "\"]"
248 c.sendOrQueue(msg)
249 }
250 251 // Send sends a raw JSON message string.
252 func (c *Conn) Send(msg string) {
253 ws.Send(c.wsConn, msg)
254 }
255 256 // Close closes the connection intentionally (no reconnect).
257 func (c *Conn) Close() {
258 c.closing = true
259 c.state = StateClosed
260 for id, sub := range c.subs {
261 sub.Filters = nil
262 sub.OnEvent = nil
263 sub.OnEOSE = nil
264 sub.conn = nil
265 delete(c.subs, id)
266 }
267 c.subs = nil
268 c.onEvent = nil
269 c.onEOSE = nil
270 c.onOK = nil
271 c.onAuth = nil
272 c.onReady = nil
273 c.ScheduleReconnect = nil
274 ws.Close(c.wsConn)
275 }
276 277 // SetOnEvent sets a global event handler (all subscriptions).
278 func (c *Conn) SetOnEvent(fn func(string, *nostr.Event)) {
279 c.onEvent = fn
280 }
281 282 // SetOnEOSE sets a global EOSE handler.
283 func (c *Conn) SetOnEOSE(fn func(string)) {
284 c.onEOSE = fn
285 }
286 287 // SetOnOK sets a handler for OK responses.
288 func (c *Conn) SetOnOK(fn func(string, bool, string)) {
289 c.onOK = fn
290 }
291 292 // SetOnAuth sets a handler for AUTH challenges.
293 func (c *Conn) SetOnAuth(fn func(string)) {
294 c.onAuth = fn
295 }
296 297 298 // sendOrQueue writes msg when the worker's socket is actually writable and
299 // queues it otherwise.
300 //
301 // The Moxie state flag and the WebSocket's readyState are set by different
302 // callbacks, so state can read "open" while the socket is still connecting;
303 // ws.Send reports that as false. Ignoring that result silently dropped the
304 // REQ, leaving the subscription absent on that relay with no error anywhere.
305 // A client-visible symptom: the feed subscription - the one with an event to
306 // deliver - never received its stored event, while subscriptions answered
307 // from the local store looked healthy.
308 func (c *Conn) sendOrQueue(msg string) {
309 if ws.ReadyState(c.wsConn) == 1 && ws.Send(c.wsConn, msg) {
310 return
311 }
312 c.queueSend(msg)
313 }
314 315 // queueSend stores msg for a later flush. The copy into the root arena is the
316 // point: msg was built in the caller's frame, and a string written into the
317 // receiver has to outlive that frame. Pushing the frame-backed string instead
318 // left pendingSend holding freed memory, and flushing it sent a zero-filled
319 // buffer of the right length - which the relay cannot parse, after which it
320 // stops answering that connection.
321 func (c *Conn) queueSend(msg string) {
322 prev := runtime.CurrentArena()
323 runtime.SovereignSetArena(runtime.RootArena())
324 cp := []byte{:len(msg)}
325 copy(cp, msg)
326 kept := string(cp)
327 runtime.SovereignRestoreArena(prev)
328 if len(c.pendingSend) >= 256 {
329 c.pendingSend = push(c.pendingSend[1:], kept)
330 return
331 }
332 c.pendingSend = push(c.pendingSend, kept)
333 }
334 335 // flushPendingSend re-sends what could not be written earlier.
336 func (c *Conn) flushPendingSend() {
337 if len(c.pendingSend) == 0 {
338 return
339 }
340 for _, m := range c.pendingSend {
341 ws.Send(c.wsConn, m)
342 }
343 c.pendingSend = c.pendingSend[:0]
344 }
345 346 // flushSubs re-sends REQ for all stored subscriptions (used after WS opens).
347 func (c *Conn) flushSubs() {
348 for _, sub := range c.subs {
349 msg := "[\"REQ\",\"" | sub.ID | "\""
350 for _, f := range sub.Filters {
351 msg |= "," | f.Serialize()
352 }
353 msg |= "]"
354 ws.Send(c.wsConn, msg)
355 }
356 }
357 358 func eventJSON(ev *nostr.Event) (s string) {
359 // Everything is appended to this buffer within its capacity: `|` does not
360 // grow, so the size has to be right before the first write. The helpers
361 // return their own complete piece, never a buffer they were handed: a
362 // returned slice comes back with len == cap (the spare capacity is not part
363 // of the value), and appending to that hands back a full slice, so the next
364 // append fails loud with "push past end of slice capacity".
365 buf := []byte{:0:jsonSize(ev)}
366 buf = buf | "{"
367 buf = buf | "\"id\":\""
368 buf = buf | escapeJSON(ev.ID)
369 buf = buf | "\",\"pubkey\":\""
370 buf = buf | escapeJSON(ev.PubKey)
371 buf = buf | "\",\"created_at\":"
372 buf = buf | itoa(ev.CreatedAt)
373 buf = buf | ",\"kind\":"
374 buf = buf | itoa(int64(ev.Kind))
375 buf = buf | ",\"tags\":"
376 buf = buf | tagsJSON(ev.Tags)
377 buf = buf | ",\"content\":\""
378 buf = buf | escapeJSON(ev.Content)
379 buf = buf | "\",\"sig\":\""
380 buf = buf | escapeJSON(ev.Sig)
381 buf = buf | "\"}"
382 return string(buf)
383 }
384 385 // jsonSize upper-bounds the serialized event: escaping can double a byte and
386 // each tag element costs its quotes plus a separator.
387 func jsonSize(ev *nostr.Event) (n int32) {
388 n = 256 + len(ev.ID)*2 + len(ev.PubKey)*2 + len(ev.Sig)*2 + len(ev.Content)*2
389 for _, tag := range ev.Tags {
390 n += 4
391 for _, el := range tag {
392 n += len(el)*2 + 4
393 }
394 }
395 return n
396 }
397 398 // tagsJSON renders the tag array whole. Its own buffer is sized from the tags,
399 // so the caller appends a finished piece rather than a half-built buffer.
400 func tagsJSON(tags [][]string) (out string) {
401 n := int32(2)
402 for _, tag := range tags {
403 n += 4
404 for _, el := range tag {
405 n += len(el)*2 + 4
406 }
407 }
408 buf := []byte{:0:n}
409 buf = buf | "["
410 for i, tag := range tags {
411 if i > 0 {
412 buf = buf | ","
413 }
414 buf = buf | "["
415 for j, s := range tag {
416 if j > 0 {
417 buf = buf | ","
418 }
419 buf = buf | "\""
420 buf = buf | escapeJSON(s)
421 buf = buf | "\""
422 }
423 buf = buf | "]"
424 }
425 buf = buf | "]"
426 return string(buf)
427 }
428 429 // escapeJSON returns the JSON string body for s, quotes excluded.
430 func escapeJSON(s string) (out string) {
431 buf := []byte{:0:len(s)*2 + 8}
432 for i := 0; i < len(s); i++ {
433 c := s[i]
434 switch c {
435 case '"':
436 buf = buf | "\\\""
437 case '\\':
438 buf = buf | "\\\\"
439 case '\n':
440 buf = buf | "\\n"
441 case '\r':
442 buf = buf | "\\r"
443 case '\t':
444 buf = buf | "\\t"
445 default:
446 buf = push(buf, c)
447 }
448 }
449 return string(buf)
450 }
451 452 func itoa(n int64) (s string) {
453 if n == 0 {
454 return "0"
455 }
456 neg := false
457 if n < 0 {
458 neg = true
459 n = -n
460 }
461 var b [20]byte
462 i := len(b)
463 for n > 0 {
464 i--
465 b[i] = byte('0' + n%10)
466 n /= 10
467 }
468 if neg {
469 i--
470 b[i] = '-'
471 }
472 return string(b[i:])
473 }
474 475 func indexOf(s string, c byte) (n int32) {
476 for i := 0; i < len(s); i++ {
477 if s[i] == c {
478 return i
479 }
480 }
481 return -1
482 }
483