837 lines
22 KiB
TypeScript
837 lines
22 KiB
TypeScript
import {derived, get, readable, writable} from "svelte/store"
|
|
import type {Readable} from "svelte/store"
|
|
import {
|
|
batch,
|
|
call,
|
|
int,
|
|
ms,
|
|
now,
|
|
on,
|
|
range,
|
|
removeUndefined,
|
|
sleep,
|
|
spec,
|
|
uniqBy,
|
|
DAY,
|
|
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 {Comment, TimeEvent} from "@welshman/domain"
|
|
import {Events} from "@welshman/app"
|
|
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 {EVENT_CONTEXT_KINDS, REACTION_KINDS} from "@app/content"
|
|
import {app, network, reader} from "@app/core"
|
|
|
|
const noEvents: TrustedEvent[] = []
|
|
|
|
const mergeSorted = <T>(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 = (event: TrustedEvent) => {
|
|
if (event.kind === COMMENT) {
|
|
const {id, address} = reader(Comment)(event).root()
|
|
|
|
return removeUndefined([id, address])
|
|
}
|
|
|
|
return [...tagValues(hexTags("e"), event.tags), ...tagValues(addressTags("a"), event.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<string[]>
|
|
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<string>()
|
|
const eventsByTarget = new Map<string, TrustedEvent[]>()
|
|
const targetsByEventId = new Map<string, string[]>()
|
|
const subscribersByTarget = new Map<string, Set<(events: TrustedEvent[]) => void>>()
|
|
const deletedChecks = new Map<string, Set<() => void>>()
|
|
|
|
const addEvent = (event: TrustedEvent, touched: Set<string>) => {
|
|
// 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<string>) => {
|
|
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<string>) => {
|
|
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<string>) => {
|
|
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<string>()
|
|
|
|
// 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<string, TrustedEvent[]>()
|
|
|
|
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<string>()
|
|
|
|
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<TrustedEvent[]> => {
|
|
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<typeof makeFeedContext>
|
|
|
|
// 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<string>) => 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<FeedLoadState>) =>
|
|
state?.status === "loading" || state?.status === "searching"
|
|
|
|
const makeFeedLoader = (load: () => Promise<FeedSpan>) => {
|
|
const state = writable<FeedLoadState>({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<FeedSpan>,
|
|
options: Partial<ScrollerOpts> = {},
|
|
) => {
|
|
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<TrustedEvent[]>([])
|
|
const seen = new Set<string>()
|
|
|
|
// Events from further back than the feed has reached, held rather than rendered as they land.
|
|
const held = new Map<string, TrustedEvent>()
|
|
|
|
// 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<TrustedEvent>) => {
|
|
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<string>) => {
|
|
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<string, {count: number; lowest: number}>()
|
|
|
|
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<FeedSpan> => {
|
|
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<number>
|
|
|
|
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<FeedSpan> => {
|
|
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 =>
|
|
app.get().use(Events).forUrl(url, filters).get().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<string>()
|
|
|
|
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)
|
|
|
|
// The filters take in comments too, for the feed context, and a TimeEvent reader throws on them.
|
|
const isValid = (event: TrustedEvent) => {
|
|
if (event.kind === EVENT_TIME) {
|
|
const timeEvent = reader(TimeEvent)(event)
|
|
|
|
return timeEvent.start() && timeEvent.end() && !seen.has(event.id)
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
const events = writable(
|
|
uniqBy(
|
|
e => e.id,
|
|
relays.flatMap(url => app.get().use(Events).forUrl(url, filters).get()),
|
|
)
|
|
.filter(spec({kind: EVENT_TIME}))
|
|
.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<string>) => {
|
|
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": [...range(since, until, DAY)].map(t => String(Math.floor(t / DAY))),
|
|
},
|
|
],
|
|
onEose: () => {
|
|
complete = true
|
|
},
|
|
})
|
|
|
|
return {found: found.length, complete}
|
|
}
|
|
|
|
const loadOlder = async (): Promise<FeedSpan> => {
|
|
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<FeedSpan> => {
|
|
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)
|
|
},
|
|
}
|
|
}
|