main.mx raw
1 package main
2
3 import (
4 "git.smesh.lol/smesh/web/common/helpers"
5 "git.smesh.lol/smesh/web/common/jsbridge/relayproxy"
6 "git.smesh.lol/smesh/web/common/jsbridge/ws"
7 "git.smesh.lol/smesh/web/common/mw"
8 "git.smesh.lol/smesh/web/common/nostr"
9 "git.smesh.lol/smesh/web/common/relay"
10 )
11
12 // Relay-proxy worker.
13 //
14 // Owns: WS pool to remote nostr relays, subscription router, MLS event
15 // subscription routing (kinds 443/445/1059 -> MLS worker via supervisor).
16 //
17 // IDB is owned by the Store Worker. This worker sends S_* messages to the
18 // supervisor which routes them to the Store Worker, and receives SR_* responses.
19 //
20 // Wire format: JSON-array MW strings.
21 // page -> worker:
22 // ["REQ", subID, filter]
23 // ["CLOSE", subID]
24 // ["PROXY", subID, filter, [relayURLs]]
25 // ["EVENT", signedEventJSON] -- publish to writeRelays
26 // ["PUBLISH_TO", signedEventJSON, [relayURLs]] -- explicit targets
27 // ["SET_PUBKEY", hex]
28 // ["SET_WRITE_RELAYS", [relayURLs]]
29 // ["ENC_KEY", hex]
30 // ["MLS_SUB", [relayURLs], [groupIDs]] -- open persistent MLS subs
31 // ["MLS_UPDATE_GROUPS", [groupIDs]] -- update kind 445 #h filter
32 // ["MLS_FETCH_KP", peer, [relayURLs]] -- one-shot key package fetch
33 // ["SR_QUERY", reqID, eventsJSON] -- response from Store Worker
34 // worker -> page:
35 // ["READY"]
36 // ["EVENT", subID, event]
37 // ["EOSE", subID]
38 // ["OK", eventID, ok, message]
39 // ["MLS_EVENT", evJSON] -- kind 443/445/1059 -> MLS worker
40 // ["MLS_KP_RESULT", peer, evJSON] -- one-shot KP fetch result -> MLS worker
41 // ["S_PUT_EVENT", evJSON] -- to Store Worker (fire-and-forget)
42 // ["S_ENC_KEY", hexKey] -- to Store Worker (fire-and-forget)
43 // ["S_QUERY", reqID, filterJSON] -- to Store Worker
44
45 // proxyEOSETimeoutMs: how long to wait for relays to EOSE before emitting
46 // EOSE to the consumer. Short value gets the consumer unblocked quickly so it
47 // can render cached data; the linger window catches stragglers.
48 var proxyEOSETimeoutMs int32 = 3000
49
50 // proxyLingerMs: after EOSE is emitted to the consumer, keep the remote subs
51 // open this long to forward late-arriving events. Solves the slow-relay race
52 // where events arrived just after the EOSE timeout fired and were dropped.
53 var proxyLingerMs int32 = 25000
54
55 type peerRelayInfo struct {
56 urls []string // relays the peer reads from (or both read+write)
57 ts int64 // created_at of the kind 10002 event
58 }
59
60 type clientSub struct {
61 filter *nostr.Filter
62 filterRaw string
63 }
64
65 type proxySub struct {
66 remoteIDs map[string]bool
67 relayCount int32
68 timer int32
69 done bool
70 live bool
71 }
72
73 var (
74 clientSubs map[string]*clientSub
75 proxySubs map[string]*proxySub
76 remoteToProxy map[string]string
77 rpool *relay.Pool
78 writeRelays []string
79 myPubkey string
80 mlsSubIDs map[string]bool // MLS event subscriptions (kinds 443/445/1059)
81 mlsKPFetchID string // in-flight one-shot KP fetch prefix
82 mlsKPFetchPeer string // peer pubkey for in-flight KP fetch
83 mlsKPFetchSubs []string // per-relay sub IDs for in-flight fetch
84 mlsGroupIDs []string // current #h filter for kind 445
85 storeCBs map[int32]func(string) // reqID -> callback for SR_QUERY responses
86 nextStoreID int32
87
88 // Peer relay list cache (kind 10002).
89 peerRelayCache map[string]*peerRelayInfo // pubkey -> cached relay list
90 rlFetchID string // in-flight relay list fetch sub prefix
91 rlFetchPeer string // peer for in-flight relay list fetch
92 rlFetchSubs []string // per-relay sub IDs
93 rlFetchCB func([]string) // callback when relay list is resolved
94 rlFetchEOSE int32 // count of EOSE received
95 )
96
97 func main() {
98 clientSubs = map[string]*clientSub{}
99 proxySubs = map[string]*proxySub{}
100 remoteToProxy = map[string]string{}
101 rpool = relay.NewPool()
102 mlsSubIDs = map[string]bool{}
103 storeCBs = map[int32]func(string){}
104 peerRelayCache = map[string]*peerRelayInfo{}
105 relayproxy.WorkerOnMessage(handleMessage)
106 relayproxy.WorkerPost(`["READY"]`)
107 }
108
109 // storeQuery sends S_QUERY to the Store Worker and calls fn with the eventsJSON response.
110 func storeQuery(filterRaw string, fn func(string)) {
111 nextStoreID++
112 reqID := nextStoreID
113 storeCBs[reqID] = fn
114 relayproxy.WorkerPost(`["S_QUERY",` | helpers.Itoa(int64(reqID)) | `,` | filterRaw | `]`)
115 rid := reqID
116 relayproxy.WorkerSetTimeout(30000, func() {
117 if _, ok := storeCBs[rid]; ok {
118 delete(storeCBs, rid)
119 }
120 })
121 }
122
123 func handleMessage(msg string) {
124 w := mw.New(msg)
125 msgType := w.Str()
126 switch msgType {
127 case "REQ":
128 subID := dup(w.Str())
129 filterRaw := dup(w.Raw())
130 handleReq(subID, filterRaw)
131 case "CLOSE":
132 subID := dup(w.Str())
133 handleClose(subID)
134 case "PROXY":
135 subID := dup(w.Str())
136 filterRaw := dup(w.Raw())
137 relayURLs := dupStrs(w.Strs())
138 handleProxy(subID, filterRaw, relayURLs, false)
139 case "PROXY_LIVE":
140 subID := dup(w.Str())
141 filterRaw := dup(w.Raw())
142 relayURLs := dupStrs(w.Strs())
143 handleProxy(subID, filterRaw, relayURLs, true)
144 case "EVENT":
145 eventRaw := w.Raw()
146 handleEventPublish(eventRaw)
147 case "PUBLISH_TO":
148 eventRaw := w.Raw()
149 relayURLs := w.Strs()
150 handlePublishTo(eventRaw, relayURLs)
151 case "SET_WRITE_RELAYS":
152 urls := dupStrs(w.Strs())
153 handleSetWriteRelays(urls)
154 case "SET_PUBKEY":
155 myPubkey = dup(w.Str())
156 case "ENC_KEY":
157 hexKey := w.Str()
158 relayproxy.WorkerPost(`["S_ENC_KEY",` | jstr(hexKey) | `]`)
159 case "SET_PROXY_EOSE_MS":
160 ms := int32(w.Num())
161 if ms > 0 {
162 proxyEOSETimeoutMs = ms
163 }
164 case "MLS_SUB":
165 urls := dupStrs(w.Strs())
166 groupIDs := dupStrs(w.Strs())
167 handleMLSSub(urls, groupIDs)
168 case "MLS_UPDATE_GROUPS":
169 groupIDs := dupStrs(w.Strs())
170 handleMLSUpdateGroups(groupIDs)
171 case "MLS_FETCH_KP":
172 peer := dup(w.Str())
173 urls := dupStrs(w.Strs())
174 handleMLSFetchKP(peer, urls)
175 // Store Worker response for storeQuery.
176 case "SR_QUERY":
177 reqID := int32(w.Num())
178 eventsJSON := w.Raw()
179 if fn, ok := storeCBs[reqID]; ok {
180 fn(eventsJSON)
181 delete(storeCBs, reqID)
182 }
183 }
184 }
185
186 func handleMLSSub(relayURLs, groupIDs []string) {
187 println("[relay-proxy] handleMLSSub: urls=" | helpers.Itoa(int64(len(relayURLs))) | " groupIDs=" | helpers.Itoa(int64(len(groupIDs))) | " pubkey=" | myPubkey[:16] | "...")
188 if myPubkey == "" || len(relayURLs) == 0 {
189 println("[relay-proxy] handleMLSSub: no pubkey or no URLs, aborting")
190 return
191 }
192 for rSubID := range mlsSubIDs {
193 for _, c := range rpool.AllConns() {
194 c.CloseSubscription(rSubID)
195 }
196 }
197 mlsSubIDs = map[string]bool{}
198 mlsGroupIDs = groupIDs
199
200 for _, url := range relayURLs {
201 url = normalizeRelayURL(url)
202 if !isAllowedRelay(url) {
203 println("[relay-proxy] handleMLSSub: blocked url=" | url)
204 continue
205 }
206 suffix := urlSuffix(url)
207 idP := "mlsp_" | suffix
208 mlsSubIDs[idP] = true
209 c := getConn(url)
210 println("[relay-proxy] handleMLSSub: sub " | idP | " kinds=[443,1059] #p=" | myPubkey[:16] | "... on " | url)
211 c.Subscribe(idP, []*nostr.Filter{{
212 Kinds: []uint32{443, 1059},
213 Tags: map[string][]string{"#p": {myPubkey}},
214 }})
215 if len(mlsGroupIDs) > 0 {
216 idH := "mlsh_" | suffix
217 mlsSubIDs[idH] = true
218 println("[relay-proxy] handleMLSSub: sub " | idH | " kind=445 #h groups=" | helpers.Itoa(int64(len(mlsGroupIDs))))
219 c.Subscribe(idH, []*nostr.Filter{{
220 Kinds: []uint32{445},
221 Tags: map[string][]string{"#h": mlsGroupIDs},
222 }})
223 }
224 }
225 }
226
227 func handleMLSUpdateGroups(groupIDs []string) {
228 mlsGroupIDs = groupIDs
229 for _, c := range rpool.AllConns() {
230 suffix := urlSuffix(c.URL)
231 idH := "mlsh_" | suffix
232 c.CloseSubscription(idH)
233 delete(mlsSubIDs, idH)
234 if len(mlsGroupIDs) > 0 && mlsSubIDs["mlsp_"|suffix] {
235 mlsSubIDs[idH] = true
236 c.Subscribe(idH, []*nostr.Filter{{
237 Kinds: []uint32{445},
238 Tags: map[string][]string{"#h": mlsGroupIDs},
239 }})
240 }
241 }
242 }
243
244 func handleMLSFetchKP(peer string, relayURLs []string) {
245 println("[relay-proxy] handleMLSFetchKP: peer=" | peer[:16] | "... urls=" | helpers.Itoa(int64(len(relayURLs))))
246 if peer == "" {
247 println("[relay-proxy] handleMLSFetchKP: no peer")
248 relayproxy.WorkerPost(`["MLS_KP_RESULT",` | jstr(peer) | `,null]`)
249 return
250 }
251 // Resolve shared relays: fetch peer's kind 10002, intersect with ours.
252 // Falls back to provided URLs if no intersection found.
253 fallback := relayURLs
254 if len(fallback) == 0 {
255 fallback = writeRelays
256 }
257 resolveSharedRelays(peer, fallback, func(urls []string) {
258 if len(urls) == 0 {
259 println("[relay-proxy] handleMLSFetchKP: no relays after resolution")
260 relayproxy.WorkerPost(`["MLS_KP_RESULT",` | jstr(peer) | `,null]`)
261 return
262 }
263 doKPFetch(peer, urls)
264 })
265 }
266
267 func doKPFetch(peer string, relayURLs []string) {
268 if mlsKPFetchID != "" {
269 closeKPFetchSubs()
270 }
271 mlsKPFetchID = "mlskp_" | peer[:8]
272 mlsKPFetchPeer = peer
273 mlsKPFetchSubs = nil
274 println("[relay-proxy] doKPFetch: peer=" | peer[:16] | "... urls=" | helpers.Itoa(int64(len(relayURLs))))
275 for _, rawURL := range relayURLs {
276 url := normalizeRelayURL(rawURL)
277 if !isAllowedRelay(url) {
278 println("[relay-proxy] doKPFetch: blocked url=" | url)
279 continue
280 }
281 doKPFetchDirect(peer, url)
282 return
283 }
284 println("[relay-proxy] doKPFetch: no valid URLs")
285 mlsKPFetchID = ""
286 relayproxy.WorkerPost(`["MLS_KP_RESULT",` | jstr(peer) | `,null]`)
287 }
288
289 // doKPFetchDirect opens a dedicated WebSocket to fetch a single KP event.
290 // Bypasses the pool connection to avoid state interference.
291 func doKPFetchDirect(peer, url string) {
292 subID := "kpq"
293 reqMsg := `["REQ","` | subID | `",{"kinds":[443],"authors":["` | peer | `"],"limit":1}]`
294 println("[relay-proxy] doKPFetchDirect: opening dedicated ws to " | url)
295 println("[relay-proxy] doKPFetchDirect: REQ=" | reqMsg)
296
297 done := false
298 fetchPeer := peer
299
300 wsConn := ws.Dial(url,
301 func(connID int32, data string) {
302 if done {
303 return
304 }
305 println("[relay-proxy] doKPFetchDirect: recv len=" | helpers.Itoa(int64(len(data))))
306 label, rSubID, payload := nostr.ParseRelayMessage(data)
307 println("[relay-proxy] doKPFetchDirect: label=" | label | " subID=" | rSubID)
308 if label == "EVENT" && rSubID == subID {
309 ev := nostr.ParseEvent(payload)
310 if ev != nil && ev.Kind == 443 {
311 done = true
312 evJSON := ev.ToJSON()
313 println("[relay-proxy] doKPFetchDirect: FOUND KP for " | fetchPeer[:16] | "...")
314 mlsKPFetchID = ""
315 ws.Close(ws.Conn(connID))
316 relayproxy.WorkerPost(`["MLS_KP_RESULT",` | jstr(fetchPeer) | `,` | evJSON | `]`)
317 }
318 }
319 if label == "EOSE" && rSubID == subID && !done {
320 done = true
321 println("[relay-proxy] doKPFetchDirect: EOSE with no KP for " | fetchPeer[:16] | "...")
322 mlsKPFetchID = ""
323 ws.Close(ws.Conn(connID))
324 relayproxy.WorkerPost(`["MLS_KP_RESULT",` | jstr(fetchPeer) | `,null]`)
325 }
326 },
327 func(connID int32) {
328 println("[relay-proxy] doKPFetchDirect: connected, sending REQ")
329 ws.Send(ws.Conn(connID), reqMsg)
330 },
331 func(connID int32, code int32, reason string) {
332 if !done {
333 done = true
334 println("[relay-proxy] doKPFetchDirect: ws closed code=" | helpers.Itoa(int64(code)))
335 mlsKPFetchID = ""
336 relayproxy.WorkerPost(`["MLS_KP_RESULT",` | jstr(fetchPeer) | `,null]`)
337 }
338 },
339 func(connID int32) {
340 if !done {
341 done = true
342 println("[relay-proxy] doKPFetchDirect: ws error")
343 mlsKPFetchID = ""
344 relayproxy.WorkerPost(`["MLS_KP_RESULT",` | jstr(fetchPeer) | `,null]`)
345 }
346 },
347 )
348 _ = wsConn
349
350 relayproxy.WorkerSetTimeout(10000, func() {
351 if !done {
352 done = true
353 println("[relay-proxy] doKPFetchDirect: TIMEOUT for " | fetchPeer[:16] | "...")
354 mlsKPFetchID = ""
355 relayproxy.WorkerPost(`["MLS_KP_RESULT",` | jstr(fetchPeer) | `,null]`)
356 }
357 })
358 }
359
360 func closeKPFetchSubs() {
361 for _, sid := range mlsKPFetchSubs {
362 for _, c := range rpool.AllConns() {
363 c.CloseSubscription(sid)
364 }
365 }
366 mlsKPFetchSubs = nil
367 }
368
369 // parseRelayList extracts relay URLs from a kind 10002 event.
370 // Returns URLs where the peer reads (or has no marker = both).
371 func parseRelayList(ev *nostr.Event) (ss []string) {
372 var urls []string
373 for _, tag := range ev.Tags {
374 if len(tag) < 2 || tag[0] != "r" {
375 continue
376 }
377 url := tag[1]
378 if len(url) < 6 {
379 continue
380 }
381 // If marker present, include only "read" or no marker (= both).
382 if len(tag) >= 3 && tag[2] == "write" {
383 continue
384 }
385 urls = append(urls, normalizeRelayURL(url))
386 }
387 return urls
388 }
389
390 // resolveSharedRelays checks the cache for the peer's relay list.
391 // If cached, returns intersection immediately via cb.
392 // If not cached, fetches kind 10002 from connected relays, then calls cb.
393 func resolveSharedRelays(peer string, fallback []string, cb func([]string)) {
394 if info, ok := peerRelayCache[peer]; ok {
395 shared := intersectRelays(writeRelays, info.urls)
396 println("[relay-proxy] resolveSharedRelays: cached peer=" | peer[:16] | "... shared=" | helpers.Itoa(int64(len(shared))))
397 if len(shared) > 0 {
398 cb(shared)
399 return
400 }
401 cb(fallback)
402 return
403 }
404 fetchPeerRelayList(peer, func(urls []string) {
405 if len(urls) > 0 {
406 peerRelayCache[peer] = &peerRelayInfo{urls: urls}
407 shared := intersectRelays(writeRelays, urls)
408 println("[relay-proxy] resolveSharedRelays: fetched peer=" | peer[:16] | "... peerURLs=" | helpers.Itoa(int64(len(urls))) | " shared=" | helpers.Itoa(int64(len(shared))))
409 if len(shared) > 0 {
410 cb(shared)
411 return
412 }
413 } else {
414 println("[relay-proxy] resolveSharedRelays: no 10002 for peer=" | peer[:16] | "... using fallback")
415 }
416 cb(fallback)
417 })
418 }
419
420 func fetchPeerRelayList(peer string, cb func([]string)) {
421 if rlFetchID != "" {
422 closeRLFetchSubs()
423 }
424 rlFetchID = "rl10k_" | peer[:8]
425 rlFetchPeer = peer
426 rlFetchSubs = nil
427 rlFetchCB = cb
428 rlFetchEOSE = 0
429
430 connURLs := rpool.URLs()
431 if len(connURLs) == 0 {
432 connURLs = writeRelays
433 }
434 for _, url := range connURLs {
435 url = normalizeRelayURL(url)
436 if !isAllowedRelay(url) {
437 continue
438 }
439 suffix := urlSuffix(url)
440 subID := rlFetchID | "_" | suffix
441 rlFetchSubs = append(rlFetchSubs, subID)
442 c := getConn(url)
443 println("[relay-proxy] fetchPeerRelayList: sub " | subID | " kind=10002 author=" | peer[:16] | "... on " | url)
444 c.Subscribe(subID, []*nostr.Filter{{
445 Kinds: []uint32{10002},
446 Authors: []string{peer},
447 Limit: 1,
448 }})
449 }
450 fetchPrefix := rlFetchID
451 relayproxy.WorkerSetTimeout(5000, func() {
452 if rlFetchID == fetchPrefix {
453 println("[relay-proxy] fetchPeerRelayList: TIMEOUT for " | rlFetchPeer[:16] | "...")
454 fn := rlFetchCB
455 closeRLFetchSubs()
456 rlFetchID = ""
457 rlFetchCB = nil
458 if fn != nil {
459 fn(nil)
460 }
461 }
462 })
463 }
464
465 func closeRLFetchSubs() {
466 for _, sid := range rlFetchSubs {
467 for _, c := range rpool.AllConns() {
468 c.CloseSubscription(sid)
469 }
470 }
471 rlFetchSubs = nil
472 }
473
474 func intersectRelays(ours, theirs []string) (ss []string) {
475 var out []string
476 for _, u := range ours {
477 for _, t := range theirs {
478 if normalizeRelayURL(u) == normalizeRelayURL(t) {
479 out = append(out, u)
480 break
481 }
482 }
483 }
484 return out
485 }
486
487 func isHex(s string) (ok bool) {
488 for i := 0; i < len(s); i++ {
489 c := s[i]
490 if !((c >= '0' && c <= '9') || (c >= 'a' && c <= 'f')) {
491 return false
492 }
493 }
494 return true
495 }
496
497 func handleEventPublish(eventRaw string) {
498 ev := nostr.ParseEvent(eventRaw)
499 if ev == nil {
500 return
501 }
502 relayproxy.WorkerPost(`["S_PUT_EVENT",` | eventRaw | `]`)
503 for _, url := range writeRelays {
504 if isAllowedRelay(url) {
505 getConn(url).Publish(ev)
506 }
507 }
508 relayproxy.WorkerPost(`["OK",` | jstr(ev.ID) | `,true,""]`)
509 }
510
511 func handlePublishTo(eventRaw string, relayURLs []string) {
512 ev := nostr.ParseEvent(eventRaw)
513 if ev == nil {
514 println("[relay-proxy] handlePublishTo: parse failed")
515 return
516 }
517 println("[relay-proxy] handlePublishTo: kind=" | helpers.Itoa(int64(ev.Kind)) | " id=" | ev.ID[:16] | "... to " | helpers.Itoa(int64(len(relayURLs))) | " relays")
518 relayproxy.WorkerPost(`["S_PUT_EVENT",` | eventRaw | `]`)
519 for _, url := range relayURLs {
520 url = normalizeRelayURL(url)
521 if isAllowedRelay(url) {
522 c := getConn(url)
523 oStr := "closed"
524 if c.IsOpen() {
525 oStr = "open"
526 }
527 println("[relay-proxy] handlePublishTo: publishing to " | url | " conn=" | oStr)
528 c.Publish(ev)
529 }
530 }
531 relayproxy.WorkerPost(`["OK",` | jstr(ev.ID) | `,true,""]`)
532 }
533
534 func handleSetWriteRelays(urls []string) {
535 writeRelays = nil
536 for _, u := range urls {
537 nu := normalizeRelayURL(u)
538 if isAllowedRelay(nu) {
539 writeRelays = append(writeRelays, nu)
540 }
541 }
542 println("[relay-proxy] SET_WRITE_RELAYS: " | helpers.Itoa(int64(len(writeRelays))) | " relays")
543 for _, u := range writeRelays {
544 println("[relay-proxy] relay: " | u)
545 }
546 }
547
548 func handleReq(subID, filterRaw string) {
549 f := nostr.ParseFilter(filterRaw)
550 if f == nil {
551 return
552 }
553 clientSubs[subID] = &clientSub{filter: f, filterRaw: filterRaw}
554 sid := subID
555 storeQuery(filterRaw, func(eventsJSON string) {
556 events := nostr.ParseEventsJSON(eventsJSON)
557 for _, ev := range events {
558 relayproxy.WorkerPost(`["EVENT",` | jstr(sid) | `,` | ev.ToJSON() | `]`)
559 }
560 relayproxy.WorkerPost(`["EOSE",` | jstr(sid) | `]`)
561 })
562 }
563
564 func handleClose(subID string) {
565 delete(clientSubs, subID)
566 cleanupProxy(subID)
567 }
568
569 func handleProxy(subID, filterRaw string, relayURLs []string, live bool) {
570 cleanupProxy(subID)
571
572 f := nostr.ParseFilter(filterRaw)
573 if f == nil {
574 return
575 }
576 clientSubs[subID] = &clientSub{filter: f, filterRaw: filterRaw}
577
578 sid := subID
579 hasSearchField := false
580 for i := 0; i+8 <= len(filterRaw); i++ {
581 if filterRaw[i:i+8] == "\"search\"" {
582 hasSearchField = true
583 break
584 }
585 }
586 if !hasSearchField {
587 storeQuery(filterRaw, func(eventsJSON string) {
588 events := nostr.ParseEventsJSON(eventsJSON)
589 for _, ev := range events {
590 relayproxy.WorkerPost(`["EVENT",` | jstr(sid) | `,` | ev.ToJSON() | `]`)
591 }
592 })
593 }
594
595 remoteIDs := map[string]bool{}
596 base := "p_" | subID | "_"
597
598 seen := map[string]bool{}
599 deduped := []string{:0:len(relayURLs)}
600 for _, url := range relayURLs {
601 url = normalizeRelayURL(url)
602 if seen[url] {
603 continue
604 }
605 seen[url] = true
606 deduped = append(deduped, url)
607 }
608
609 proxySubs[subID] = &proxySub{
610 remoteIDs: remoteIDs,
611 relayCount: len(deduped),
612 live: live,
613 }
614
615 for _, url := range deduped {
616 if !isAllowedRelay(url) {
617 continue
618 }
619 suffix := urlSuffix(url)
620 rSubID := base | suffix
621 remoteIDs[rSubID] = true
622 remoteToProxy[rSubID] = subID
623 c := getConn(url)
624 c.Subscribe(rSubID, []*nostr.Filter{f})
625 }
626
627 proxyID := subID
628 proxySubs[subID].timer = relayproxy.WorkerSetTimeout(proxyEOSETimeoutMs, func() {
629 info, ok := proxySubs[proxyID]
630 if !ok || info.done {
631 return
632 }
633 info.done = true
634 if _, ok := clientSubs[proxyID]; ok {
635 relayproxy.WorkerPost(`["EOSE",` | jstr(proxyID) | `]`)
636 }
637 if info.live {
638 return
639 }
640 // Linger: keep the remote subs open for proxyLingerMs after EOSE so
641 // events that arrive late from slow relays still get forwarded. The
642 // consumer already got its EOSE so it can proceed, but stragglers
643 // arriving during the linger window still match findProxySub and
644 // reach the app. Closing immediately on EOSE timer was the source of
645 // inconsistent profile resolution.
646 info.timer = relayproxy.WorkerSetTimeout(proxyLingerMs, func() {
647 info, ok := proxySubs[proxyID]
648 if !ok {
649 return
650 }
651 for rSubID := range info.remoteIDs {
652 delete(remoteToProxy, rSubID)
653 for _, c := range rpool.AllConns() {
654 c.CloseSubscription(rSubID)
655 }
656 }
657 delete(proxySubs, proxyID)
658 delete(clientSubs, proxyID)
659 })
660 })
661 }
662
663 func cleanupProxy(proxyID string) {
664 info, ok := proxySubs[proxyID]
665 if !ok {
666 return
667 }
668 relayproxy.WorkerClearTimeout(info.timer)
669
670 if !info.done {
671 if _, ok := clientSubs[proxyID]; ok {
672 relayproxy.WorkerPost(`["EOSE",` | jstr(proxyID) | `]`)
673 }
674 }
675
676 for rSubID := range info.remoteIDs {
677 delete(remoteToProxy, rSubID)
678 for _, c := range rpool.AllConns() {
679 c.CloseSubscription(rSubID)
680 }
681 }
682 delete(proxySubs, proxyID)
683 }
684
685 func getConn(url string) (p *relay.Conn) {
686 c := rpool.Connect(url)
687 if c.ScheduleReconnect == nil {
688 wireConn(c)
689 }
690 return c
691 }
692
693 func wireConn(c *relay.Conn) {
694 url := c.URL
695 c.ScheduleReconnect = func(delayMs int32, fn func()) {
696 relayproxy.WorkerSetTimeout(delayMs, fn)
697 }
698 c.SetOnEvent(func(rSubID string, ev *nostr.Event) {
699 if mlsSubIDs[rSubID] {
700 evJSON := ev.ToJSON()
701 println("[relay-proxy] MLS_EVENT: sub=" | rSubID | " kind=" | helpers.Itoa(int64(ev.Kind)) | " from=" | ev.PubKey[:16] | "...")
702 relayproxy.WorkerPost(`["MLS_EVENT",` | evJSON | `]`)
703 return
704 }
705 if mlsKPFetchID != "" && len(rSubID) >= len(mlsKPFetchID) && rSubID[:len(mlsKPFetchID)] == mlsKPFetchID && ev.Kind == 443 {
706 evJSON := ev.ToJSON()
707 peer := mlsKPFetchPeer
708 println("[relay-proxy] MLS_KP_RESULT: found KP for " | peer[:16] | "... on sub=" | rSubID)
709 closeKPFetchSubs()
710 mlsKPFetchID = ""
711 relayproxy.WorkerPost(`["MLS_KP_RESULT",` | jstr(peer) | `,` | evJSON | `]`)
712 return
713 }
714 if rlFetchID != "" && len(rSubID) >= len(rlFetchID) && rSubID[:len(rlFetchID)] == rlFetchID && ev.Kind == 10002 {
715 urls := parseRelayList(ev)
716 peer := rlFetchPeer
717 println("[relay-proxy] RL_RESULT: peer=" | peer[:16] | "... urls=" | helpers.Itoa(int64(len(urls))))
718 fn := rlFetchCB
719 closeRLFetchSubs()
720 rlFetchID = ""
721 rlFetchCB = nil
722 if fn != nil {
723 fn(urls)
724 }
725 return
726 }
727 // Proxy event path: skip CheckSig here - the Verify Supervisor in
728 // wasm-host.mjs intercepts ["EVENT",...] and verifies before forwarding
729 // to the app worker. Unverified events never reach app business logic.
730 proxyID, info := findProxySub(rSubID)
731 if info == nil {
732 return
733 }
734 cs, ok := clientSubs[proxyID]
735 if !ok {
736 return
737 }
738 if cs.filter != nil && !cs.filter.Matches(ev) {
739 return
740 }
741 evJSON := ev.ToJSON()
742 relayproxy.WorkerPost(`["S_PUT_EVENT",` | evJSON | `]`)
743 relayproxy.WorkerPost(`["EVENT",` | jstr(proxyID) | `,` | evJSON | `]`)
744 relayproxy.WorkerPost(`["SEEN_ON",` | jstr(ev.ID) | `,` | jstr(url) | `]`)
745 })
746 c.SetOnEOSE(func(rSubID string) {
747 if mlsSubIDs[rSubID] || (mlsKPFetchID != "" && len(rSubID) >= len(mlsKPFetchID) && rSubID[:len(mlsKPFetchID)] == mlsKPFetchID) {
748 println("[relay-proxy] MLS EOSE: sub=" | rSubID | " relay=" | url)
749 }
750 if rlFetchID != "" && len(rSubID) >= len(rlFetchID) && rSubID[:len(rlFetchID)] == rlFetchID {
751 rlFetchEOSE++
752 println("[relay-proxy] RL EOSE: sub=" | rSubID | " relay=" | url | " eose=" | helpers.Itoa(int64(rlFetchEOSE)) | "/" | helpers.Itoa(int64(len(rlFetchSubs))))
753 if rlFetchEOSE >= len(rlFetchSubs) {
754 fn := rlFetchCB
755 closeRLFetchSubs()
756 rlFetchID = ""
757 rlFetchCB = nil
758 if fn != nil {
759 fn(nil)
760 }
761 }
762 }
763 })
764 c.SetOnOK(func(eventID string, ok bool, msg string) {
765 okStr := "true"
766 if !ok {
767 okStr = "false"
768 println("[relay-proxy] OK REJECTED: id=" | eventID[:16] | "... ok=" | okStr | " msg=" | msg | " relay=" | url)
769 }
770 relayproxy.WorkerPost(`["OK",` | jstr(eventID) | `,` | okStr | `,` | jstr(msg) | `]`)
771 if ok {
772 relayproxy.WorkerPost(`["SEEN_ON",` | jstr(eventID) | `,` | jstr(url) | `]`)
773 }
774 })
775 }
776
777 func findProxySub(rSubID string) (string, *proxySub) {
778 proxyID, ok := remoteToProxy[rSubID]
779 if !ok {
780 return "", nil
781 }
782 info, ok := proxySubs[proxyID]
783 if !ok {
784 delete(remoteToProxy, rSubID)
785 return "", nil
786 }
787 return proxyID, info
788 }
789
790 func normalizeRelayURL(url string) (s string) {
791 for len(url) > 0 && url[len(url)-1] == '/' {
792 url = url[:len(url)-1]
793 }
794 return url
795 }
796
797 func isAllowedRelay(url string) (ok bool) {
798 if len(url) >= 6 && url[:6] == "wss://" {
799 return true
800 }
801 if len(url) >= 16 && url[:16] == "ws://localhost:" {
802 return true
803 }
804 if len(url) >= 15 && url[:15] == "ws://127.0.0.1:" {
805 return true
806 }
807 return false
808 }
809
810 func urlSuffix(url string) (s string) {
811 // Extract host from wss://host/ or wss://host
812 start := 0
813 for i := 0; i < len(url); i++ {
814 if url[i] == '/' && i+1 < len(url) && url[i+1] == '/' {
815 start = i + 2
816 break
817 }
818 }
819 end := len(url)
820 for i := start; i < len(url); i++ {
821 if url[i] == '/' || url[i] == ':' {
822 end = i
823 break
824 }
825 }
826 host := url[start:end]
827 out := []byte{:0:len(host)}
828 for i := 0; i < len(host); i++ {
829 c := host[i]
830 if (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') {
831 out = append(out, c)
832 }
833 }
834 if len(out) > 12 {
835 out = out[len(out)-12:]
836 }
837 return string(out)
838 }
839
840 func jstr(s string) (sv string) { return helpers.JsonString(s) }
841
842 func dup(s string) (sv string) {
843 if len(s) == 0 {
844 return ""
845 }
846 b := []byte{:len(s)}
847 copy(b, s)
848 return string(b)
849 }
850
851 func dupStrs(ss []string) (ss2 []string) {
852 if len(ss) == 0 {
853 return nil
854 }
855 out := []string{:0:len(ss)}
856 for _, s := range ss {
857 out = append(out, dup(s))
858 }
859 return out
860 }
861