import {derived, get, readable, writable} from "svelte/store" import type {Readable} from "svelte/store" import {batch, call, int, ms, now, on, sleep, uniqBy, MONTH, YEAR} from "@welshman/lib" import { COMMENT, DELETE, EVENT_TIME, NOTE, addressTags, compareEventsAsc, getAddress, getCommentFiltersForRoot, getIdOrAddress, getReplyFilters, hexTags, isReplaceableKind, matchFilters, tagValues, } from "@welshman/util" import {TimeEvent} from "@welshman/domain" import type {Maybe} from "@welshman/lib" import type {Filter, TrustedEvent} from "@welshman/util" import {mergeRepositoryUpdates} from "@welshman/net" import type {RepositoryUpdate} from "@welshman/net" import {createScroller} from "@lib/html" import type {ScrollerOpts} from "@lib/html" import {daysBetween} from "@lib/util" import {EVENT_CONTEXT_KINDS, REACTION_KINDS} from "@app/content" import {app, network, reader} from "@app/core" import {getEventsForUrl} from "@app/repository" const noEvents: TrustedEvent[] = [] const mergeSorted = (left: T[], right: T[], compare: (a: T, b: T) => number) => { const merged: T[] = [] let i = 0 let j = 0 while (i < left.length && j < right.length) { merged.push(compare(left[i], right[j]) <= 0 ? left[i++] : right[j++]) } while (i < left.length) { merged.push(left[i++]) } while (j < right.length) { merged.push(right[j++]) } return merged } // A NIP-22 comment points at its thread root with `E`/`A`, which is the scope a reply count wants. const getTargets = ({kind, tags}: TrustedEvent) => kind === COMMENT ? [...tagValues(hexTags("E"), tags), ...tagValues(addressTags("A"), tags)] : [...tagValues(hexTags("e"), tags), ...tagValues(addressTags("a"), tags)] // A related event may point at a replaceable one by either id or address, so both are keys const getKeys = (event: TrustedEvent) => isReplaceableKind(event.kind) ? [event.id, getAddress(event)] : [event.id] export const makeFeedContext = ({ relays, withReplies = false, }: { relays: string[] | Promise withReplies?: boolean }) => { // Kind 1 notes reply with `e` tags rather than as NIP-22 comments. const contextKinds = withReplies ? [...EVENT_CONTEXT_KINDS, NOTE] : EVENT_CONTEXT_KINDS const requestKinds = withReplies ? [...REACTION_KINDS, NOTE] : REACTION_KINDS const {repository, tracker} = app.get() const controller = new AbortController() const targets = new Set() const eventsByTarget = new Map() const targetsByEventId = new Map() const subscribersByTarget = new Map void>>() const deletedChecks = new Map void>>() const addEvent = (event: TrustedEvent, touched: Set) => { // An event seen before its target stays unfiled, so adding the target picks it up from the repository. if (!targetsByEventId.has(event.id)) { const eventTargets = getTargets(event).filter(target => targets.has(target)) if (eventTargets.length > 0) { targetsByEventId.set(event.id, eventTargets) for (const target of eventTargets) { eventsByTarget.set(target, [...(eventsByTarget.get(target) || []), event]) touched.add(target) } } } } const removeEvent = (id: string, touched: Set) => { const eventTargets = targetsByEventId.get(id) if (eventTargets) { targetsByEventId.delete(id) for (const target of eventTargets) { const events = eventsByTarget.get(target)!.filter(event => event.id !== id) if (events.length > 0) { eventsByTarget.set(target, events) } else { eventsByTarget.delete(target) } touched.add(target) } } } const notifyDeleted = (added: TrustedEvent[], removed: Set) => { if (deletedChecks.size > 0) { if (removed.size > 0) { for (const checks of deletedChecks.values()) { for (const check of checks) { check() } } } else { for (const event of added) { for (const check of deletedChecks.get(getIdOrAddress(event)) || []) { check() } } } } } const notify = (touched: Set) => { for (const target of touched) { for (const subscriber of subscribersByTarget.get(target) || []) { subscriber(eventsByTarget.get(target) || noEvents) } } } const loadFrom = async (urls: string[], events: TrustedEvent[]) => { const context = await network.get().loadComplete({ relays: urls, signal: controller.signal, filters: [ ...getReplyFilters(events, {kinds: requestKinds}), ...getCommentFiltersForRoot(events), ], }) if (context.length > 0) { network.get().loadLenient({ relays: urls, signal: controller.signal, filters: getReplyFilters(context, {kinds: [DELETE]}), }) } } const loadContext = batch(100, async (events: TrustedEvent[]) => { const touched = new Set() // Local events never come through the update listener, so file them before asking the network. for (const event of repository.query([ ...getReplyFilters(events, {kinds: contextKinds}), ...getCommentFiltersForRoot(events), ])) { addEvent(event, touched) } notify(touched) const urls = await relays // A feed built out of other people's outboxes has no relay holding the whole conversation. const eventsByRelay = new Map() for (const event of events) { for (const url of tracker.getRelays(event.id)) { if (!urls.includes(url)) { eventsByRelay.set(url, [...(eventsByRelay.get(url) || []), event]) } } } await Promise.all([ loadFrom(urls, events), ...Array.from(eventsByRelay, ([url, seenThere]) => loadFrom([url], seenThere)), ]) }) const unsubscribe = on( repository, "update", batch(150, (updates: RepositoryUpdate[]) => { const {added, removed} = mergeRepositoryUpdates(updates) const touched = new Set() for (const event of added) { if (contextKinds.includes(event.kind)) { addEvent(event, touched) } } for (const id of removed) { removeEvent(id, touched) } notify(touched) notifyDeleted(added, removed) }), ) // Track an event so its context gets requested with the rest of the batch const add = (event: TrustedEvent) => { if (!targets.has(event.id)) { for (const key of getKeys(event)) { targets.add(key) } loadContext(event) } } const relatedForKey = (key: string) => readable(eventsByTarget.get(key) || noEvents, set => { let subscribers = subscribersByTarget.get(key) if (!subscribers) { subscribers = new Set() subscribersByTarget.set(key, subscribers) } subscribers.add(set) set(eventsByTarget.get(key) || noEvents) return () => { subscribers.delete(set) if (subscribers.size === 0) { subscribersByTarget.delete(key) } } }) return { add, deleted: (event: TrustedEvent) => readable(repository.isDeleted(event), set => { const key = getIdOrAddress(event) const check = () => set(repository.isDeleted(event)) let checks = deletedChecks.get(key) if (!checks) { checks = new Set() deletedChecks.set(key, checks) } checks.add(check) check() return () => { checks.delete(check) if (checks.size === 0) { deletedChecks.delete(key) } } }), related: (event: TrustedEvent): Readable => { add(event) const [key, ...rest] = getKeys(event) // A replaceable event collects both buckets, and something tagging it by id and address lands in both. return rest.length > 0 ? derived([relatedForKey(key), ...rest.map(relatedForKey)], buckets => uniqBy(event => event.id, buckets.flat()), ) : relatedForKey(key) }, cleanup: () => { controller.abort() unsubscribe() }, } } export type FeedContext = ReturnType // Keeps a feed's store in step with the repository: later arrivals, deletions, and events newly seen. const syncFeed = ({ relays, filters, addEvents, removeEvents, requireRelay = true, }: { relays: string[] filters: Filter[] addEvents: (events: TrustedEvent[]) => void removeEvents: (ids: Set) => void // Whether an event has to have been seen on one of `relays` to belong to this feed requireRelay?: boolean }) => { const onTrackedId = batch(150, (ids: string[]) => { const matching: TrustedEvent[] = [] for (const id of new Set(ids)) { const event = app.get().repository.getEvent(id) if (event && matchFilters(filters, event)) { matching.push(event) } } if (matching.length > 0) { addEvents(matching) } }) return [ on( app.get().repository, "update", batch(150, (updates: RepositoryUpdate[]) => { const {added, removed} = mergeRepositoryUpdates(updates) if (removed.size > 0) { removeEvents(removed) } const matching = added.filter( event => matchFilters(filters, event) && (!requireRelay || relays.some(url => app.get().tracker.getRelays(event.id).has(url))), ) if (matching.length > 0) { addEvents(matching) } }), ), on(app.get().tracker, "add", (id: string, url: string) => { if (relays.includes(url)) { onTrackedId(id) } }), ] } // One direction of a feed. An empty span is a gap in the timeline rather than the end of it. export type FeedLoadState = | {status: "idle"} | {status: "loading"} | {status: "searching"} | {status: "exhausted"} // One span of a feed's timeline. `complete` is whether the relays answered, which a dropped request is not. export type FeedSpan = {found: number; complete: boolean; exhausted: boolean} // Relatively high because quiet relays eose first. const SPAN_THRESHOLD = 0.8 // Short because a relay can accept a socket and never eose. const SPAN_TIMEOUT = 3000 // Abort resolves a request with what arrived. const spanSignal = (signal: AbortSignal) => AbortSignal.any([signal, AbortSignal.timeout(SPAN_TIMEOUT)]) // Enough to cross a gap, not enough to reach the end of the history in one trigger. const SPANS_PER_TRIGGER = 3 // A month-wide span is thousands of events in a busy feed, so ask for a page and let relays answer from the anchor. const PAGE_SIZE = 100 // A span that finds nothing widens the next one, so a sparse history takes fewer round trips. const nextInterval = (interval: number, found: number) => found > 0 ? int(MONTH) : Math.round(interval * 1.5) // Whether a request is in flight, which is not whether more might exist. export const isFeedLoading = (state: Maybe) => state?.status === "loading" || state?.status === "searching" const makeFeedLoader = (load: () => Promise) => { const state = writable({status: "idle"}) let running = false const run = async () => { if (running || get(state).status === "exhausted") { return } running = true try { for (let span = 0; span < SPANS_PER_TRIGGER; span++) { state.set({status: span > 0 ? "searching" : "loading"}) const {found, complete, exhausted} = await load() if (exhausted) { state.set({status: "exhausted"}) return } // A span nobody answered for has not moved the window, so hold here rather than walking past it. if (!complete) { await sleep(ms(3)) break } if (found > 0) { break } } } finally { running = false } } // A run covers a few spans, so settling at the end of one blinks the spinner once per page. const settle = () => { if (!running && isFeedLoading(get(state))) { state.set({status: "idle"}) } } return {subscribe: state.subscribe, run, settle} } // The container's orientation decides which direction `reverse` reaches, so the caller passes it. export const makeScrollLoader = ( element: HTMLElement, load: () => Promise, options: Partial = {}, ) => { const loader = makeFeedLoader(load) const scroller = createScroller({ element, delay: 300, threshold: 5000, ...options, onScroll: loader.run, onSettle: loader.settle, }) return {subscribe: loader.subscribe, stop: scroller.stop} } // Holds what a view has loaded and knows how to ask for the next span, while the view decides when. export const makeFeed = ({ relays, filters, onEvent, at = now(), }: { relays: string[] filters: Filter[] onEvent?: (event: TrustedEvent) => void at?: number }) => { const controller = new AbortController() const events = writable([]) const seen = new Set() // Events from further back than the feed has reached, held rather than rendered as they land. const held = new Map() // Each direction widens its own span, since what one walks through says nothing about the other. let oldest = at let newest = at let olderInterval = int(MONTH) let newerInterval = int(MONTH) // How far back the feed has been answered for, which decides whether an event is ready to render. let reached = at // A span reaches only as far as its shallowest relay, so every one of them has to have answered. const relayCount = new Set(relays).size const insertEvents = (newEvents: Iterable) => { const added: TrustedEvent[] = [] for (const event of newEvents) { if (!seen.has(event.id)) { seen.add(event.id) held.delete(event.id) added.push(event) onEvent?.(event) } } if (added.length > 0) { added.sort(compareEventsAsc) events.update($events => mergeSorted($events, added, compareEventsAsc)) } } const addEvents = (newEvents: TrustedEvent[]) => { const ready: TrustedEvent[] = [] for (const event of newEvents) { if (!seen.has(event.id) && !held.has(event.id)) { if (event.created_at >= reached) { ready.push(event) } else { held.set(event.id, event) } } } insertEvents(ready) } const removeEvents = (ids: Set) => { events.update($events => $events.filter(event => !ids.has(event.id))) for (const id of ids) { seen.delete(id) held.delete(id) } } // Take the feed back to `timestamp`, releasing everything held for the stretch it gains. const reach = (timestamp: number) => { if (timestamp < reached) { reached = timestamp const ready: TrustedEvent[] = [] for (const [id, event] of held) { if (event.created_at >= reached) { held.delete(id) ready.push(event) } } insertEvents(ready) } } const unsubscribers = syncFeed({relays, filters, addEvents, removeEvents}) // Each relay answers a limit for itself, so a page runs out where the relay that gave least ran out. const loadSpan = async (extension: Filter, onCover?: (timestamp: number) => void) => { let complete = false const pages = new Map() const countEvent = (event: TrustedEvent, url: string) => { const page = pages.get(url) if (page) { page.count += 1 page.lowest = Math.min(page.lowest, event.created_at) } else { pages.set(url, {count: 1, lowest: event.created_at}) } if (pages.size === relayCount) { onCover?.(Math.max(...Array.from(pages.values(), page => page.lowest))) } } const found = await network.get().request({ relays, autoClose: true, // The socket sends five per 100ms first come first served, so the feed on screen goes ahead of the space's background pulls. priority: 1, signal: spanSignal(controller.signal), threshold: SPAN_THRESHOLD, filters: filters.map(filter => ({...filter, ...extension})), onEvent: countEvent, onDuplicate: countEvent, onEose: () => { complete = true }, }) return {found, complete, pages} } // The window only moves once the relays have answered, since a dropped request looks like an empty span. const loadOlder = async (): Promise => { if (oldest < now() - int(2, YEAR)) { return {found: 0, complete: true, exhausted: true} } const until = oldest const since = until - olderInterval // A page comes back newest first, so a relay's lowest answer proves the stretch above it. const {found, complete, pages} = await loadSpan({since, until, limit: PAGE_SIZE}, reach) // A relay that answered with less than its limit has covered its whole span. let edge: Maybe for (const page of pages.values()) { if (page.count >= PAGE_SIZE && (edge === undefined || page.lowest > edge)) { edge = page.lowest } } if (complete) { olderInterval = nextInterval(olderInterval, found.length) // Stepping the edge back is what stops a page that filled up inside one span being asked for again. oldest = edge === undefined ? since : Math.min(edge, until - 1) } // addEvents, not insertEvents: found can reach below oldest. addEvents(found) reach(oldest) return {found: found.length, complete, exhausted: false} } // A limit is answered with the newest events matching it, which reaches away from an anchor in the past. const loadNewer = async (): Promise => { if (newest >= now()) { return {found: 0, complete: true, exhausted: true} } const since = newest const until = Math.min(now(), since + newerInterval) const {found, complete} = await loadSpan({since, until}) if (complete) { newerInterval = nextInterval(newerInterval, found.length) newest = until } addEvents(found) return {found: found.length, complete, exhausted: false} } // What the repository already holds is an answer too, so a page of it renders before any relay replies. const stored = relays.map(url => Array.from(getEventsForUrl(url, filters)).sort(compareEventsAsc).slice(-PAGE_SIZE), ) if (stored.every(page => page.length > 0)) { reach(Math.max(...stored.map(page => page[0].created_at))) } addEvents(stored.flat()) return { events, loadOlder, loadNewer, cleanup: () => { controller.abort() unsubscribers.forEach(call) }, } } // Calendar events are addressed by the days they cover, so the spans are date hashes. export const makeCalendarFeed = ({ relays, filters, onEvent, }: { relays: string[] filters: Filter[] onEvent?: (event: TrustedEvent) => void }) => { const interval = int(5, MONTH) const controller = new AbortController() const seen = new Set() let oldest = now() let newest = now() const getStart = (event: TrustedEvent) => reader(TimeEvent)(event).start() ?? 0 const compareByStart = (a: TrustedEvent, b: TrustedEvent) => getStart(a) - getStart(b) || compareEventsAsc(a, b) const isValid = (event: TrustedEvent) => { const timeEvent = reader(TimeEvent)(event) return timeEvent.start() && timeEvent.end() && !seen.has(event.id) } const events = writable( uniqBy( e => e.id, relays.flatMap(url => Array.from(getEventsForUrl(url, filters))), ).sort(compareByStart), ) const insertEvents = (newEvents: TrustedEvent[]) => { const valid = newEvents.filter(isValid) if (valid.length === 0) { return } for (const event of valid) { seen.add(event.id) onEvent?.(event) } valid.sort(compareByStart) events.update($events => { // Calendar events are addressable, so a new version supersedes the old one const superseded = new Set(valid.map(getAddress)) return mergeSorted( $events.filter(e => !superseded.has(getAddress(e))), valid, compareByStart, ) }) } const removeEvents = (ids: Set) => { events.update($events => $events.filter(event => !ids.has(event.id))) for (const id of ids) { seen.delete(id) } } // Calendar events are addressable and often relayed on, so this takes any match rather than its own relays'. const unsubscribers = syncFeed({ relays, filters, addEvents: insertEvents, removeEvents, requireRelay: false, }) const loadTimeframe = async (since: number, until: number) => { let complete = false const found = await network.get().request({ relays, autoClose: true, priority: 1, signal: spanSignal(controller.signal), threshold: SPAN_THRESHOLD, filters: [{kinds: [EVENT_TIME], "#D": daysBetween(since, until).map(String)}], onEose: () => { complete = true }, }) return {found: found.length, complete} } const loadOlder = async (): Promise => { if (oldest < now() - int(2, YEAR)) { return {found: 0, complete: true, exhausted: true} } const until = oldest const since = until - interval const {found, complete} = await loadTimeframe(since, until) if (complete) { oldest = since } return {found, complete, exhausted: false} } const loadNewer = async (): Promise => { if (newest > now() + int(2, YEAR)) { return {found: 0, complete: true, exhausted: true} } const since = newest const until = since + interval const {found, complete} = await loadTimeframe(since, until) if (complete) { newest = until } return {found, complete, exhausted: false} } return { events, loadOlder, loadNewer, // The month and week views jump to a range rather than scrolling, and wait to show progress for it. load: loadTimeframe, cleanup: () => { controller.abort() unsubscribers.forEach(call) }, } }