main.mx raw
1 package main
2
3 // Profile Worker - owns all author profile data and subscription management.
4 //
5 // Data authority: authorNames/Pics/Content/Ts/Relays/Follows/Mutes (all profiles).
6 // Shell keeps: authorNames/Pics view cache (small, for sync render), followSet/muteSet
7 // (sync filter derived sets), myFollows/myMutes (own lists for action operations).
8 //
9 // Wire: supervisor <-> worker
10 // supervisor -> worker:
11 // ["P_RESOLVE", pk] -- request fetch if not already fetched
12 // ["P_RESOLVE_FORCE", pk] -- re-fetch unconditionally
13 // ["P_EVENT", evJSON] -- kind 0/3/10002/10000 event from ap-* or prof sub
14 // ["P_EOSE", subID] -- ap-* EOSE (retry logic)
15 // ["P_SET_RELAYS", urlsJSON]
16 // ["P_SET_PUBKEY", pk]
17 // ["P_IDB_LOADED", pksJSON] -- Shell finished IDB load; mark these pks fetched
18 //
19 // worker -> supervisor (Shell or relay-proxy):
20 // ["P_RESOLVED", pk, name, pic] -- kind 0 resolved
21 // ["P_CONTENT", pk, contentJSON] -- full kind 0 content
22 // ["P_FOLLOWS", pk, pksJSON] -- kind 3 follow list
23 // ["P_MUTES", pk, pksJSON] -- kind 10000 mute list
24 // ["P_RELAYS", pk, urlsJSON] -- kind 10002 relay list
25 // ["P_SUB", subID, filterJSON, urlsJSON]
26 // ["P_CLOSE", subID]
27 // ["P_RETRY_REQ"] -- request Shell's list of still-missing pks
28
29 import (
30 "git.smesh.lol/smesh/web/common/helpers"
31 "git.smesh.lol/smesh/web/common/jsbridge/profile"
32 "git.smesh.lol/smesh/web/common/mw"
33 "git.smesh.lol/smesh/web/common/nostr"
34 )
35
36 var (
37 // Operational state (fetch queue management)
38 fetchedK0 map[string]bool
39 fetchedK10k map[string]bool
40 resolvedSet map[string]bool // pk -> kind 0 resolved with a name
41 requestedSet map[string]int32 // pk -> retry attempts so far (capped at retryMax)
42 authorSubPK map[string]string
43 fetchQueue []string
44 fetchTimer int32
45 retryRound int32
46 retryTimer int32
47 sweepTimer int32
48 subCounter int32
49 relayURLs []string
50 myPK string
51
52 // Relay discovery (two-gen rotation with author caches)
53 authorRelays map[string][]string
54 authorRelaysOld map[string][]string
55 relayFreq map[string]int32
56 relayFreqOld map[string]int32
57
58 // Profile data (canonical authority - two-generation rotation)
59 authorNames map[string]string
60 authorNamesOld map[string]string
61 authorPics map[string]string
62 authorPicsOld map[string]string
63 authorContent map[string]string
64 authorContentOld map[string]string
65 authorTs map[string]int64
66 authorTsOld map[string]int64
67 authorFollows map[string][]string
68 authorMutes map[string][]string
69 authorCacheCount int32
70 )
71
72 const rotateAuthorMax = 500
73
74 // retryMax is the per-pk attempt cap. Each attempt waits sweepIntervalMs.
75 const retryMax = 6
76 const sweepIntervalMs = 15000
77
78 var discoveryRelays = []string{
79 "wss://purplepag.es",
80 "wss://relay.damus.io",
81 "wss://nos.lol",
82 }
83
84 func main() {
85 fetchedK0 = map[string]bool{}
86 fetchedK10k = map[string]bool{}
87 resolvedSet = map[string]bool{}
88 requestedSet = map[string]int32{}
89 authorSubPK = map[string]string{}
90 authorRelays = map[string][]string{}
91 authorRelaysOld = map[string][]string{}
92 relayFreq = map[string]int32{}
93 relayFreqOld = map[string]int32{}
94 authorNames = map[string]string{}
95 authorNamesOld = map[string]string{}
96 authorPics = map[string]string{}
97 authorPicsOld = map[string]string{}
98 authorContent = map[string]string{}
99 authorContentOld = map[string]string{}
100 authorTs = map[string]int64{}
101 authorTsOld = map[string]int64{}
102 authorFollows = map[string][]string{}
103 authorMutes = map[string][]string{}
104 profile.WorkerOnMessage(handleMessage)
105 scheduleSweep()
106 }
107
108 // scheduleSweep arms the periodic retry timer. Every sweepIntervalMs the
109 // worker re-fetches any pks in requestedSet that aren't in resolvedSet, until
110 // each pk hits retryMax. This is independent of EOSE-based retry and works
111 // even when subs were closed without producing the expected event.
112 func scheduleSweep() {
113 if sweepTimer != 0 {
114 profile.WorkerClearTimeout(sweepTimer)
115 }
116 sweepTimer = profile.WorkerSetTimeout(int32(sweepIntervalMs), func() {
117 sweepTimer = 0
118 runSweep()
119 scheduleSweep()
120 })
121 }
122
123 func runSweep() {
124 var missing []string
125 for pk, attempts := range requestedSet {
126 if resolvedSet[pk] {
127 continue
128 }
129 if attempts >= retryMax {
130 continue
131 }
132 missing = append(missing, pk)
133 }
134 if len(missing) == 0 {
135 return
136 }
137 for _, pk := range missing {
138 requestedSet[pk] = requestedSet[pk] + 1
139 fetchedK0[pk] = false
140 }
141 proxy := buildProxyURLs(topRelays(8))
142 const batchSize = 100
143 for i := 0; i < len(missing); i += batchSize {
144 end := i + batchSize
145 if end > len(missing) {
146 end = len(missing)
147 }
148 chunk := missing[i:end]
149 authors := buildAuthorsJSON(chunk)
150 limit := helpers.Itoa(int64(len(chunk) * 2))
151 subCounter++
152 emitSub("ap-sweep-" | helpers.Itoa(int64(subCounter)),
153 `{"authors":` | authors | `,"kinds":[0,10002],"limit":` | limit | `}`, proxy)
154 subCounter++
155 emitSub("ap-d-" | helpers.Itoa(int64(subCounter)),
156 `{"authors":` | authors | `,"kinds":[0],"limit":` | helpers.Itoa(int64(len(chunk))) | `}`, relayURLs)
157 }
158 }
159
160 func rotateAuthorCaches() {
161 authorNamesOld = authorNames
162 authorNames = map[string]string{}
163 authorPicsOld = authorPics
164 authorPics = map[string]string{}
165 authorContentOld = authorContent
166 authorContent = map[string]string{}
167 authorTsOld = authorTs
168 authorTs = map[string]int64{}
169 authorRelaysOld = authorRelays
170 authorRelays = map[string][]string{}
171 relayFreqOld = relayFreq
172 relayFreq = map[string]int32{}
173 // Clear fetch tracking maps on rotation. If we kept these populated while
174 // the data caches were thrown out, the worker would refuse to re-fetch
175 // authors whose data was rotated away ("we already fetched X, but the
176 // cache is empty for X" → permanent blank).
177 fetchedK0 = map[string]bool{}
178 fetchedK10k = map[string]bool{}
179 resolvedSet = map[string]bool{}
180 requestedSet = map[string]int32{}
181 authorCacheCount = 0
182 }
183
184 func handleMessage(msg string) {
185 w := mw.New(msg)
186 switch w.Str() {
187 case "P_RESOLVE":
188 pk := w.Str()
189 queueProfileFetch(pk)
190 case "P_RESOLVE_FORCE":
191 pk := w.Str()
192 fetchedK0[pk] = false
193 queueProfileFetch(pk)
194 case "P_EVENT":
195 evJSON := w.Raw()
196 ev := nostr.ParseEvent(evJSON)
197 if ev != nil {
198 applyEvent(ev)
199 }
200 case "P_EOSE":
201 subID := w.Str()
202 handleEOSE(subID)
203 case "P_SET_RELAYS":
204 relayURLs = parseStringArray(w.Raw())
205 case "P_SET_PUBKEY":
206 myPK = w.Str()
207 case "P_IDB_LOADED":
208 // Shell finished loading profiles from smesh-kv IDB. Mark them fetched AND
209 // resolved so Profile Worker doesn't waste relay bandwidth re-fetching them.
210 pksJSON := w.Raw()
211 pks := parseStringArray(pksJSON)
212 for _, pk := range pks {
213 fetchedK0[pk] = true
214 resolvedSet[pk] = true
215 }
216 case "P_RETRY_PROVIDE":
217 // Shell sent list of pending pubkeys without resolved names. Re-queue
218 // them, the periodic sweep + queue flush will pick them up. We bypass
219 // per-pk attempt cap here because this is a user-driven retry signal
220 // (tab switch, refresh button) not an automatic background retry.
221 pksJSON := w.Raw()
222 pks := parseStringArray(pksJSON)
223 for _, pk := range pks {
224 if resolvedSet[pk] {
225 continue
226 }
227 fetchedK0[pk] = false
228 if _, ok := requestedSet[pk]; ok {
229 requestedSet[pk] = 0 // reset attempts on user-driven retry
230 } else {
231 requestedSet[pk] = 0
232 }
233 queueProfileFetch(pk)
234 }
235 }
236 }
237
238 func applyEvent(ev *nostr.Event) {
239 switch ev.Kind {
240 case 0:
241 applyKind0(ev)
242 case 3:
243 applyKind3(ev)
244 case 10000:
245 applyKind10000(ev)
246 case 10002:
247 applyKind10002(ev)
248 }
249 }
250
251 func applyKind0(ev *nostr.Event) {
252 prevTs, hasPrev := authorTs[ev.PubKey]
253 if hasPrev && ev.CreatedAt <= prevTs {
254 return
255 }
256 if _, exists := authorContent[ev.PubKey]; !exists {
257 authorCacheCount++
258 if authorCacheCount > rotateAuthorMax {
259 rotateAuthorCaches()
260 }
261 }
262 authorTs[ev.PubKey] = ev.CreatedAt
263 authorContent[ev.PubKey] = ev.Content
264
265 name := helpers.JsonGetString(ev.Content, "name")
266 if len(name) == 0 {
267 name = helpers.JsonGetString(ev.Content, "display_name")
268 }
269 pic := helpers.JsonGetString(ev.Content, "picture")
270 if len(name) > 0 {
271 authorNames[ev.PubKey] = name
272 resolvedSet[ev.PubKey] = true
273 }
274 if len(pic) > 0 {
275 authorPics[ev.PubKey] = pic
276 }
277
278 // Emit P_RESOLVED so Shell can fill note header placeholders.
279 profile.WorkerPost(`["P_RESOLVED",` | jstr(ev.PubKey) | `,` | jstr(name) | `,` | jstr(pic) | `]`)
280 // Emit P_CONTENT so Shell can update profile page and zap buttons.
281 if len(ev.Content) > 0 {
282 profile.WorkerPost(`["P_CONTENT",` | jstr(ev.PubKey) | `,` | helpers.JsonString(ev.Content) | `]`)
283 }
284 }
285
286 func applyKind3(ev *nostr.Event) {
287 var pks []string
288 for _, tag := range ev.Tags.GetAll("p") {
289 if v := tag.Value(); v != "" {
290 pks = append(pks, v)
291 }
292 }
293 authorFollows[ev.PubKey] = pks
294 profile.WorkerPost(`["P_FOLLOWS",` | jstr(ev.PubKey) | `,` | buildAuthorsJSON(pks) | `]`)
295 }
296
297 func applyKind10000(ev *nostr.Event) {
298 var pks []string
299 for _, tag := range ev.Tags.GetAll("p") {
300 if v := tag.Value(); v != "" {
301 pks = append(pks, v)
302 }
303 }
304 authorMutes[ev.PubKey] = pks
305 profile.WorkerPost(`["P_MUTES",` | jstr(ev.PubKey) | `,` | buildAuthorsJSON(pks) | `]`)
306 }
307
308 func applyKind10002(ev *nostr.Event) {
309 tags := ev.Tags.GetAll("r")
310 if tags == nil {
311 return
312 }
313 var urls []string
314 for _, tag := range tags {
315 u := string(tag.Value())
316 if u != "" {
317 urls = append(urls, u)
318 relayFreq[u] = relayFreq[u] + 1
319 }
320 }
321 if len(urls) > 0 {
322 authorRelays[ev.PubKey] = urls
323 profile.WorkerPost(`["P_RELAYS",` | jstr(ev.PubKey) | `,` | buildURLsJSON(urls) | `]`)
324 }
325 }
326
327 func queueProfileFetch(pk string) {
328 if len(pk) != 64 {
329 return
330 }
331 // Track every requested pk so the periodic sweep can retry until
332 // resolved or per-pk attempt cap is hit.
333 if _, ok := requestedSet[pk]; !ok {
334 requestedSet[pk] = 0
335 }
336 if fetchedK0[pk] {
337 return
338 }
339 fetchedK0[pk] = true
340 fetchQueue = append(fetchQueue, pk)
341 if fetchTimer != 0 {
342 profile.WorkerClearTimeout(fetchTimer)
343 }
344 fetchTimer = profile.WorkerSetTimeout(300, func() {
345 fetchTimer = 0
346 flushFetchQueue()
347 })
348 }
349
350 func flushFetchQueue() {
351 if len(fetchQueue) == 0 {
352 return
353 }
354 queue := fetchQueue
355 fetchQueue = nil
356 proxy := buildProxyURLs(nil)
357 const batchSize = 100
358 for i := 0; i < len(queue); i += batchSize {
359 end := i + batchSize
360 if end > len(queue) {
361 end = len(queue)
362 }
363 chunk := queue[i:end]
364 authors := buildAuthorsJSON(chunk)
365 limit := helpers.Itoa(int64(len(chunk)))
366 subCounter++
367 emitSub("ap-batch-q-" | helpers.Itoa(int64(subCounter)),
368 `{"authors":` | authors | `,"kinds":[0],"limit":` | limit | `}`, proxy)
369 subCounter++
370 emitSub("ap-d-" | helpers.Itoa(int64(subCounter)),
371 `{"authors":` | authors | `,"kinds":[0],"limit":` | limit | `}`, relayURLs)
372 }
373 }
374
375 func fetchAuthorProfile(pk string) {
376 if fetchedK0[pk] {
377 return
378 }
379 fetchedK0[pk] = true
380 subCounter++
381 subID := "ap-" | helpers.Itoa(int64(subCounter))
382 authorSubPK[subID] = pk
383 emitSub(subID, `{"authors":[` | jstr(pk) | `],"kinds":[0,3,10002,10000],"limit":6}`, buildProxy(pk))
384 }
385
386 func handleEOSE(subID string) {
387 // "ap-batch-" prefix matches both "ap-batch-q-N" (queue flush) and
388 // "ap-batch-N" (legacy P_RETRY_PROVIDE re-batch). The periodic sweep
389 // now handles per-pk retries with its own cap, so EOSE-triggered retry
390 // is no longer needed. Keeping it here would double-fetch and the
391 // global retryRound cap caused unresolved pks to permanently stick.
392 if len(subID) > 9 && subID[:9] == "ap-batch-" {
393 return
394 }
395 if len(subID) > 9 && subID[:9] == "ap-sweep-" {
396 return
397 }
398 if len(subID) > 3 && subID[:3] == "ap-" {
399 pk, ok := authorSubPK[subID]
400 if !ok {
401 return
402 }
403 delete(authorSubPK, subID)
404 if !resolvedSet[pk] {
405 rels, ok2 := authorRelays[pk]
406 if !ok2 {
407 rels, ok2 = authorRelaysOld[pk]
408 }
409 if ok2 && len(rels) > 0 && !fetchedK10k[pk] {
410 fetchedK10k[pk] = true
411 fetchedK0[pk] = false
412 fetchAuthorProfile(pk)
413 }
414 }
415 }
416 }
417
418 func emitSub(subID, filterJSON string, urls []string) {
419 profile.WorkerPost(`["P_SUB",` | jstr(subID) | `,` | filterJSON | `,` | buildURLsJSON(urls) | `]`)
420 }
421
422 func buildProxy(pk string) (ss []string) {
423 out := []string{:len(discoveryRelays)}
424 copy(out, discoveryRelays)
425 for _, u := range relayURLs {
426 out = appendUnique(out, u)
427 }
428 if rels, ok := authorRelays[pk]; ok {
429 for _, r := range rels {
430 out = appendUnique(out, r)
431 }
432 } else if rels, ok := authorRelaysOld[pk]; ok {
433 for _, r := range rels {
434 out = appendUnique(out, r)
435 }
436 }
437 for _, r := range topRelays(4) {
438 out = appendUnique(out, r)
439 }
440 return out
441 }
442
443 func buildProxyURLs(extra []string) (ss []string) {
444 out := []string{:len(discoveryRelays)}
445 copy(out, discoveryRelays)
446 for _, u := range relayURLs {
447 out = appendUnique(out, u)
448 }
449 for _, u := range extra {
450 out = appendUnique(out, u)
451 }
452 return out
453 }
454
455 func topRelays(n int32) (ss []string) {
456 type kv struct{ url string; count int32 }
457 var all []kv
458 for url, count := range relayFreq {
459 all = append(all, kv{url, count})
460 }
461 for i := 0; i < len(all); i++ {
462 for j := i + 1; j < len(all); j++ {
463 if all[j].count > all[i].count {
464 all[i], all[j] = all[j], all[i]
465 }
466 }
467 }
468 var result []string
469 for i := 0; i < len(all) && i < n; i++ {
470 result = append(result, all[i].url)
471 }
472 return result
473 }
474
475 func appendUnique(list []string, val string) (ss []string) {
476 for _, v := range list {
477 if v == val {
478 return list
479 }
480 }
481 return append(list, val)
482 }
483
484 func buildAuthorsJSON(pks []string) (s string) {
485 s := "["
486 for i, pk := range pks {
487 if i > 0 { s = s | "," }
488 s = s | jstr(pk)
489 }
490 return s | "]"
491 }
492
493 func buildURLsJSON(urls []string) (s string) {
494 s := "["
495 for i, u := range urls {
496 if i > 0 { s = s | "," }
497 s = s | jstr(u)
498 }
499 return s | "]"
500 }
501
502 func parseStringArray(json string) (ss []string) {
503 var out []string
504 i := 0
505 for i < len(json) && json[i] != '[' { i++ }
506 if i >= len(json) { return nil }
507 i++
508 for {
509 for i < len(json) && (json[i] == ' ' || json[i] == ',' || json[i] == '\n') { i++ }
510 if i >= len(json) || json[i] == ']' { break }
511 if json[i] != '"' { break }
512 i++
513 start := i
514 for i < len(json) && json[i] != '"' {
515 if json[i] == '\\' { i++ }
516 i++
517 }
518 if i >= len(json) { break }
519 out = append(out, json[start:i])
520 i++
521 }
522 return out
523 }
524
525 func jstr(s string) (sv string) { return helpers.JsonString(s) }
526
527