timeline.actor.ts raw
1 import { Func, Inbox, Lifecycle, spawn, loop, STOP } from 'anneal'
2 import { Filter, Event as NEvent, matchFilters } from 'nostr-tools'
3 import { SimplePool } from 'nostr-tools'
4 import { AbstractRelay } from 'nostr-tools/abstract-relay'
5 import { sha256 } from '@noble/hashes/sha2'
6 import dayjs from 'dayjs'
7 import { ISigner, TSubRequestFilter } from '@/types'
8 import { EventCacheActor } from './event-cache.actor'
9 import { createSubscription } from './subscription.actor'
10 import { TTimelineRef, TimelineData, SubscribeReq, SubHandle, LoadMoreReq } from './types'
11
12 type SubState = {
13 events: NEvent[]
14 eosedAt: number | null
15 cachedEvents: NEvent[]
16 filter: TSubRequestFilter
17 urls: string[]
18 onEvents: (events: NEvent[], eosed: boolean) => void
19 onNew: (evt: NEvent) => void
20 needSort: boolean
21 subCloser: () => void
22 flushTimer: ReturnType<typeof setTimeout> | null
23 lastFlushedCount: number
24 }
25
26 export type TimelineActorConfig = {
27 eventCacheActor: EventCacheActor
28 pool: SimplePool
29 getSigner: () => ISigner | undefined
30 trackEventSeenOn: (id: string, relay: AbstractRelay) => void
31 addEventToCache: (event: NEvent) => void
32 }
33
34 export type TimelineActor = {
35 subscribe: Func<SubscribeReq, SubHandle>
36 loadMore: Func<LoadMoreReq, NEvent[]>
37 injectEvent: Inbox<NEvent>
38 lc: Lifecycle
39 }
40
41 function generateTimelineKey(urls: string[], filter: Filter): string {
42 const stableFilter: Record<string, unknown> = {}
43 Object.entries(filter)
44 .sort()
45 .forEach(([key, value]) => {
46 if (key === 'limit') return
47 if (Array.isArray(value)) {
48 stableFilter[key] = [...value].sort()
49 }
50 stableFilter[key] = value
51 })
52 const paramsStr = JSON.stringify({
53 urls: [...urls].sort(),
54 filter: stableFilter
55 })
56 const data = new TextEncoder().encode(paramsStr)
57 const hashBuffer = sha256(data)
58 return Array.from(new Uint8Array(hashBuffer))
59 .map((b) => b.toString(16).padStart(2, '0'))
60 .join('')
61 }
62
63 export function createTimelineActor(config: TimelineActorConfig): TimelineActor {
64 const timelines: Record<string, TimelineData | undefined> = {}
65 const activeSubs = new Map<string, SubState>()
66
67 const subscribeCh = new Func<SubscribeReq, SubHandle>()
68 const loadMoreCh = new Func<LoadMoreReq, NEvent[]>()
69 const eventArrived = new Inbox<{ key: string; event: NEvent }>(0)
70 const eoseArrived = new Inbox<{ key: string; eosed: boolean }>(0)
71 const flushRequested = new Inbox<string>(0)
72 const injectEvent = new Inbox<NEvent>(0)
73 const lc = new Lifecycle()
74
75 function processEvent(key: string, event: NEvent, sub: SubState) {
76 config.addEventToCache(event)
77
78 if (sub.eosedAt === null) {
79 sub.events.push(event)
80 if (sub.flushTimer === null) {
81 sub.flushTimer = setTimeout(() => flushRequested.send(key), 150)
82 }
83 return
84 }
85
86 if (event.created_at > sub.eosedAt) {
87 sub.onNew(event)
88 }
89
90 const timeline = timelines[key]
91 if (!timeline || !timeline.refs.length) return
92
93 let idx = 0
94 for (const ref of timeline.refs) {
95 if (
96 event.created_at > ref[1] ||
97 (event.created_at === ref[1] && event.id < ref[0])
98 ) {
99 break
100 }
101 if (event.created_at === ref[1] && event.id === ref[0]) return
102 idx++
103 }
104 if (idx >= timeline.refs.length) return
105
106 timeline.refs.splice(idx, 0, [event.id, event.created_at])
107 }
108
109 spawn(lc, async () => {
110 await loop(
111 lc.on(() => {
112 for (const [, sub] of activeSubs) {
113 if (sub.flushTimer !== null) clearTimeout(sub.flushTimer)
114 sub.subCloser()
115 }
116 return STOP
117 }),
118
119 subscribeCh.on(async ({ req, reply }) => {
120 const { urls, filter, onEvents, onNew, startLogin, needSort = true } = req
121 const relays = Array.from(new Set(urls))
122 const key = generateTimelineKey(relays, filter)
123
124 let cachedEvents: NEvent[] = []
125 let since: number | undefined
126
127 const existingTimeline = timelines[key]
128 if (existingTimeline && existingTimeline.refs.length && needSort) {
129 const ids = existingTimeline.refs.slice(0, filter.limit).map(([id]) => id)
130 const looked = await config.eventCacheActor.lookupMany.call(ids)
131 cachedEvents = looked.filter((evt): evt is NEvent => evt != null)
132 if (cachedEvents.length) {
133 onEvents([...cachedEvents], false)
134 since = cachedEvents[0].created_at + 1
135 }
136 }
137
138 const adjustedFilter: Filter = since ? { ...filter, since } : filter
139
140 const sub = createSubscription({
141 urls: relays,
142 filters: [adjustedFilter],
143 pool: config.pool,
144 signer: config.getSigner(),
145 startLogin,
146 onevent: (evt) => eventArrived.send({ key, event: evt }),
147 oneose: (eosed) => eoseArrived.send({ key, eosed }),
148 onReceived: config.trackEventSeenOn
149 })
150
151 const subState: SubState = {
152 events: [],
153 eosedAt: null,
154 cachedEvents,
155 filter,
156 urls: relays,
157 onEvents,
158 onNew,
159 needSort,
160 subCloser: () => sub.close(),
161 flushTimer: null,
162 lastFlushedCount: 0
163 }
164 activeSubs.set(key, subState)
165
166 reply({
167 timelineKey: key,
168 closer: () => {
169 if (subState.flushTimer !== null) {
170 clearTimeout(subState.flushTimer)
171 subState.flushTimer = null
172 }
173 subState.onEvents = () => {}
174 subState.onNew = () => {}
175 sub.close()
176 activeSubs.delete(key)
177 }
178 })
179 }),
180
181 eventArrived.on(({ key, event }) => {
182 const sub = activeSubs.get(key)
183 if (!sub) return
184 processEvent(key, event, sub)
185 }),
186
187 flushRequested.on((key) => {
188 const sub = activeSubs.get(key)
189 if (!sub) return
190 sub.flushTimer = null
191 if (sub.eosedAt !== null || sub.events.length === sub.lastFlushedCount) return
192 sub.lastFlushedCount = sub.events.length
193 const sorted = sub.needSort
194 ? [...sub.events].sort((a, b) => b.created_at - a.created_at).slice(0, sub.filter.limit)
195 : [...sub.events]
196 const merged = sub.cachedEvents.length > 0
197 ? sorted.concat(sub.cachedEvents).slice(0, sub.filter.limit)
198 : sorted
199 sub.onEvents(merged, false)
200 }),
201
202 eoseArrived.on(({ key, eosed }) => {
203 const sub = activeSubs.get(key)
204 if (!sub) return
205
206 if (sub.flushTimer !== null) {
207 clearTimeout(sub.flushTimer)
208 sub.flushTimer = null
209 }
210
211 if (eosed && sub.eosedAt === null) {
212 sub.eosedAt = dayjs().unix()
213 }
214
215 if (!sub.needSort) {
216 sub.onEvents([...sub.events], sub.eosedAt !== null)
217 return
218 }
219
220 if (!eosed) {
221 sub.events = sub.events
222 .sort((a, b) => b.created_at - a.created_at)
223 .slice(0, sub.filter.limit)
224 sub.onEvents(
225 [...sub.events.concat(sub.cachedEvents).slice(0, sub.filter.limit)],
226 false
227 )
228 return
229 }
230
231 sub.events = sub.events
232 .sort((a, b) => b.created_at - a.created_at)
233 .slice(0, sub.filter.limit)
234
235 const timeline = timelines[key]
236
237 if (!timeline || !timeline.refs.length) {
238 timelines[key] = {
239 refs: sub.events.map(
240 (evt) => [evt.id, evt.created_at] as TTimelineRef
241 ),
242 filter: sub.filter,
243 urls: sub.urls
244 }
245 sub.onEvents([...sub.events], true)
246 return
247 }
248
249 const firstRefCreatedAt = timeline.refs[0][1]
250 const newRefs = sub.events
251 .filter((evt) => evt.created_at > firstRefCreatedAt)
252 .map((evt) => [evt.id, evt.created_at] as TTimelineRef)
253
254 if (sub.events.length >= sub.filter.limit) {
255 timeline.refs = newRefs
256 sub.onEvents([...sub.events], true)
257 } else {
258 timeline.refs = newRefs.concat(timeline.refs)
259 sub.onEvents(
260 [
261 ...sub.events
262 .concat(sub.cachedEvents)
263 .slice(0, sub.filter.limit)
264 ],
265 true
266 )
267 }
268 }),
269
270 injectEvent.on((event) => {
271 for (const [key, sub] of activeSubs) {
272 if (!matchFilters([sub.filter], event)) continue
273 processEvent(key, event, sub)
274 }
275 }),
276
277 loadMoreCh.on(async ({ req: { key, until, limit }, reply }) => {
278 const timeline = timelines[key]
279 if (!timeline) {
280 reply([])
281 return
282 }
283
284 const { filter, urls, refs } = timeline
285 const startIdx = refs.findIndex(([, createdAt]) => createdAt <= until)
286 const cachedIds =
287 startIdx >= 0
288 ? refs.slice(startIdx, startIdx + limit).map(([id]) => id)
289 : []
290 const cachedEvents =
291 cachedIds.length > 0
292 ? ((await config.eventCacheActor.lookupMany.call(cachedIds)).filter(
293 (e): e is NEvent => e != null
294 ) as NEvent[])
295 : []
296
297 if (cachedEvents.length >= limit) {
298 reply(cachedEvents)
299 return
300 }
301
302 const adjustedUntil = cachedEvents.length
303 ? cachedEvents[cachedEvents.length - 1].created_at - 1
304 : until
305 const adjustedLimit = limit - cachedEvents.length
306
307 const queryEvents = await new Promise<NEvent[]>((resolve) => {
308 const events: NEvent[] = []
309 const sub = createSubscription({
310 urls,
311 filters: [{ ...filter, until: adjustedUntil, limit: adjustedLimit }],
312 pool: config.pool,
313 signer: config.getSigner(),
314 onevent: (evt) => {
315 config.addEventToCache(evt)
316 events.push(evt)
317 },
318 oneose: (eosed) => {
319 if (eosed) {
320 sub.close()
321 resolve(events)
322 }
323 },
324 onclose: () => resolve(events)
325 })
326 })
327
328 const sorted = queryEvents
329 .sort((a, b) => b.created_at - a.created_at)
330 .slice(0, adjustedLimit)
331
332 const lastRefCreatedAt =
333 refs.length > 0 ? refs[refs.length - 1][1] : dayjs().unix()
334 refs.push(
335 ...sorted
336 .filter((evt) => evt.created_at < lastRefCreatedAt)
337 .map((evt) => [evt.id, evt.created_at] as TTimelineRef)
338 )
339
340 reply([...cachedEvents, ...sorted])
341 })
342 )
343 })
344
345 return { subscribe: subscribeCh, loadMore: loadMoreCh, injectEvent, lc }
346 }
347