main.mx raw
1 package main
2
3 // Feed Worker - subscription management, dedup, filtering, and pagination for the note feed.
4 //
5 // Shell forwards verified feed events (subID "feed" / "feed-more") to this worker.
6 // This worker deduplicates, applies mute/follow filters, and sends render commands back to Shell.
7 //
8 // supervisor -> worker (from Shell via feed.Send):
9 // ["F_EVENT", subID, evJSON] -- feed or feed-more event arrived
10 // ["F_EOSE", subID] -- EOSE for feed or feed-more sub
11 // ["F_SET_MODE", mode] -- "follows", "relays", or relay URL
12 // ["F_SET_RELAYS", urlsJSON]
13 // ["F_SET_FOLLOWS", pksJSON] -- current follow list (pubkeys)
14 // ["F_SET_MUTES", pksJSON] -- current mute list (pubkeys)
15 // ["F_SET_PUBKEY", pk]
16 // ["F_LOAD_MORE"] -- user clicked "load more"
17 // ["F_REFRESH"] -- user triggered refresh
18 //
19 // worker -> supervisor (to Shell via F_* prefix routing):
20 // ["F_RENDER", evJSON, now] -- now=1: render immediately, 0: buffer
21 // ["F_NEW"] -- new buffered events available (pulse)
22 // ["F_EOSE_DONE", subID] -- signal Shell to update UI after EOSE
23 // ["F_STATUS", count, exhausted] -- event count and exhaustion state
24 // ["F_SUB", subID, filterJSON, urlsJSON]
25 // ["F_CLOSE", subID]
26
27 import (
28 "runtime"
29 "git.smesh.lol/musiquay/web/common/accum"
30 "git.smesh.lol/musiquay/web/common/helpers"
31 "git.smesh.lol/musiquay/web/common/jsbridge/feed"
32 "git.smesh.lol/musiquay/web/common/mw"
33 "git.smesh.lol/nostr/pkg/core"
34 )
35
36
37 // feedState is the feed
38 // Package globals are immutable outside initialization, so it lives in
39 // one self-mutating type reached through a package-level pointer.
40 type feedState struct {
41 feedMode string
42 relayURLs accum.Strings
43 followList accum.Strings
44 followSet map[string]bool
45 muteSet map[string]bool
46 myPK string
47 seenEvents map[string]bool
48 oldestFeedTs int64
49 eventCount int32
50 feedInitialLoad bool
51 feedExhausted bool
52 feedEmptyStreak int32
53 feedLoading bool
54 feedMoreGot int32
55 feedMoreTimer int32
56 }
57
58 var feedSt *feedState
59
60 func initState() {
61 if feedSt != nil {
62 return
63 }
64 // feedSt and everything it holds live as long as this worker: build them
65 // in the root arena, not in this call's frame arena which dies on
66 // return.
67 runtime.SovereignSetArena(runtime.RootArena())
68 feedSt = &feedState{}
69 feedSt.seenEvents = map[string]bool{}
70 feedSt.followSet = map[string]bool{}
71 feedSt.muteSet = map[string]bool{}
72 feedSt.feedInitialLoad = true
73 runtime.SovereignRestoreArena(runtime.RootArena())
74 }
75
76 func main() {
77 initState()
78 feed.WorkerOnMessage(handleMessage)
79 }
80
81 func handleMessage(msg string) {
82 w := mw.New(msg)
83 switch w.Str() {
84 case "F_EVENT":
85 subID := w.Str()
86 evJSON := w.Raw()
87 handleFeedEvent(subID, evJSON)
88 case "F_EOSE":
89 subID := w.Str()
90 handleEOSE(subID)
91 case "F_SET_MODE":
92 feedSt.feedMode = w.Str()
93 case "F_SET_RELAYS":
94 feedSt.relayURLs.Set(parseStringArray(w.Raw()))
95 case "F_SET_FOLLOWS":
96 feedSt.followList.Set(parseStringArray(w.Raw()))
97 feedSt.followSet = map[string]bool{}
98 for i := int32(0); i < feedSt.followList.Len(); i++ {
99 feedSt.followSet[feedSt.followList.At(i)] = true
100 }
101 case "F_SET_MUTES":
102 pks := parseStringArray(w.Raw())
103 feedSt.muteSet = map[string]bool{}
104 for _, pk := range pks {
105 feedSt.muteSet[pk] = true
106 }
107 case "F_SET_PUBKEY":
108 feedSt.myPK = w.Str()
109 case "F_LOAD_MORE":
110 handleLoadMore()
111 case "F_REFRESH":
112 handleRefresh()
113 }
114 }
115
116 func handleFeedEvent(subID, evJSON string) {
117 ev := nostr.ParseEvent(evJSON)
118 if ev == nil {
119 return
120 }
121 if feedSt.seenEvents[ev.ID] {
122 return
123 }
124 feedSt.seenEvents[ev.ID] = true
125 if !feedPassesFilter(ev) {
126 return
127 }
128 if feedSt.muteSet[ev.PubKey] {
129 return
130 }
131 if repliesToMuted(ev) {
132 return
133 }
134 if (ev.Kind == 1 || ev.Kind == 1111) && looksLikeJSONSpam(ev.Content) {
135 return
136 }
137 if feedSt.oldestFeedTs == 0 || ev.CreatedAt < feedSt.oldestFeedTs {
138 feedSt.oldestFeedTs = ev.CreatedAt
139 }
140 feedSt.eventCount++
141 if subID == "feed-more" {
142 feedSt.feedMoreGot++
143 feed.WorkerPost(`["F_RENDER",` | evJSON | `,1]`)
144 return
145 }
146 // Events for the initial feed subscription are the feed: render them
147 // immediately. feedSt.feedInitialLoad cannot decide this, because the feed has two
148 // event sources (the local cache query and the relay subscription) and
149 // whichever EOSE arrives first clears the flag: when the local query's
150 // empty EOSE won the race, the relay's stored events arrived afterwards,
151 // were buffered as "new posts", and the feed looked empty.
152 feed.WorkerPost(`["F_RENDER",` | evJSON | `,1]`)
153 }
154
155 func handleEOSE(subID string) {
156 if subID == "feed" {
157 feedSt.feedInitialLoad = false
158 feed.WorkerPost(`["F_EOSE_DONE","feed"]`)
159 feed.WorkerPost(`["F_STATUS",` | helpers.Itoa(int64(feedSt.eventCount)) | `,0]`)
160 } else if subID == "feed-more" {
161 if feedSt.feedMoreTimer != 0 {
162 feed.WorkerClearTimeout(feedSt.feedMoreTimer)
163 feedSt.feedMoreTimer = 0
164 }
165 feedSt.feedLoading = false
166 if feedSt.feedMoreGot == 0 {
167 feedSt.feedEmptyStreak++
168 if feedSt.feedEmptyStreak >= 3 {
169 feedSt.feedExhausted = true
170 }
171 } else {
172 feedSt.feedEmptyStreak = 0
173 }
174 ex := int64(0)
175 if feedSt.feedExhausted {
176 ex = 1
177 }
178 feed.WorkerPost(`["F_STATUS",` | helpers.Itoa(int64(feedSt.eventCount)) | `,` | helpers.Itoa(ex) | `]`)
179 feed.WorkerPost(`["F_EOSE_DONE","feed-more"]`)
180 feed.WorkerPost(`["F_CLOSE","feed-more"]`)
181 }
182 }
183
184 func handleRefresh() {
185 feedSt.seenEvents = map[string]bool{}
186 feedSt.oldestFeedTs = 0
187 feedSt.eventCount = 0
188 feedSt.feedInitialLoad = true
189 feedSt.feedExhausted = false
190 feedSt.feedEmptyStreak = 0
191 feed.WorkerPost(`["F_CLOSE","feed"]`)
192 feed.WorkerPost(`["F_CLOSE","feed-more"]`)
193 subscribe()
194 }
195
196 func handleLoadMore() {
197 if feedSt.feedLoading || feedSt.feedExhausted || feedSt.oldestFeedTs == 0 {
198 return
199 }
200 feedSt.feedLoading = true
201 feedSt.feedMoreGot = 0
202 filter := buildFeedFilter(20)
203 until := `,"until":` | helpers.Itoa(feedSt.oldestFeedTs) | `}`
204 filter = filter[:len(filter)-1] | until
205 urls := feedRelays()
206 feed.WorkerPost(`["F_SUB","feed-more",` | filter | `,` | buildURLsJSON(urls) | `]`)
207 feedSt.feedMoreTimer = feed.WorkerSetTimeout(10000, func() {
208 feedSt.feedMoreTimer = 0
209 feedSt.feedLoading = false
210 feed.WorkerPost(`["F_EOSE_DONE","feed-more"]`)
211 })
212 }
213
214 func subscribe() {
215 if feedSt.myPK == "" || feedSt.relayURLs.Len() == 0 {
216 return
217 }
218 filter := buildFeedFilter(20)
219 urls := feedRelays()
220 feed.WorkerPost(`["F_SUB","feed",` | filter | `,` | buildURLsJSON(urls) | `]`)
221 }
222
223 func buildFeedFilter(limit int32) (s string) {
224 if feedSt.feedMode == "follows" && feedSt.followList.Len() > 0 {
225 authors := jstr(feedSt.myPK)
226 for i := int32(0); i < feedSt.followList.Len(); i++ {
227 pk := feedSt.followList.At(i)
228 if pk != feedSt.myPK {
229 authors = authors | "," | jstr(pk)
230 }
231 }
232 return `{"kinds":[1,6,7,1111],"authors":[` | authors | `],"limit":` | helpers.Itoa(int64(limit)) | `}`
233 }
234 return `{"kinds":[1,6,7,1111],"limit":` | helpers.Itoa(int64(limit)) | `}`
235 }
236
237 func feedPassesFilter(ev *nostr.Event) (ok bool) {
238 if feedSt.feedMode != "follows" {
239 return true
240 }
241 if feedSt.followList.Len() == 0 {
242 return true
243 }
244 if ev.PubKey == feedSt.myPK {
245 return true
246 }
247 return feedSt.followSet[ev.PubKey]
248 }
249
250 func feedRelays() (ss []string) {
251 if feedSt.feedMode != "" && feedSt.feedMode != "follows" && feedSt.feedMode != "relays" {
252 return []string{feedSt.feedMode}
253 }
254 return feedSt.relayURLs.Slice()
255 }
256
257 func repliesToMuted(ev *nostr.Event) (ok bool) {
258 for _, tag := range nostr.TagsGetAll(ev.Tags, "p") {
259 if v := string(nostr.TagValue(tag)); v != "" && feedSt.muteSet[v] {
260 return true
261 }
262 }
263 return false
264 }
265
266 func looksLikeJSONSpam(content string) (ok bool) {
267 i := 0
268 for i < len(content) {
269 c := content[i]
270 if c == ' ' || c == '\n' || c == '\r' || c == '\t' {
271 i++
272 } else {
273 break
274 }
275 }
276 if i >= len(content) {
277 return false
278 }
279 open := content[i]
280 return open == '{' || open == '['
281 }
282
283 func buildURLsJSON(urls []string) (s string) {
284 s := "["
285 for i, u := range urls {
286 if i > 0 { s = s | "," }
287 s = s | jstr(u)
288 }
289 return s | "]"
290 }
291
292 func parseStringArray(json string) (ss []string) {
293 var out []string
294 i := 0
295 for i < len(json) && json[i] != '[' { i++ }
296 if i >= len(json) { return nil }
297 i++
298 for {
299 for i < len(json) && (json[i] == ' ' || json[i] == ',' || json[i] == '\n') { i++ }
300 if i >= len(json) || json[i] == ']' { break }
301 if json[i] != '"' { break }
302 i++
303 start := i
304 for i < len(json) && json[i] != '"' {
305 if json[i] == '\\' { i++ }
306 i++
307 }
308 if i >= len(json) { break }
309 out = push(out, json[start:i])
310 i++
311 }
312 return out
313 }
314
315 func jstr(s string) (sv string) { return helpers.JsonString(s) }
316