refact status processing

src/components/Queue.svelte

@@ -3,8 +3,8 @@
 
     {#if $next}
         <div class="entry" on:click={() => select($next)}>
-            <div class="title">{$next.metadata.title}</div>
+            <div class="title">{$next.title}</div>
-            <div class="user">by {$next.status.account.acct}</div>
+            <div class="user">by {$next.username}</div>
         </div>
     {/if}
 
@@ -15,10 +15,10 @@
 
     <h6>HISTORY</h6>
 
-    {#each history as track (track.status.id)}
+    {#each history as track}
         <div class="entry" class:active={track === $current} on:click={() => select(track)}>
-            <div class>{track.metadata.title}</div>
+            <div class>{track.title}</div>
-            <div class>shared by {track.status.account.acct}</div>
+            <div class>shared by {track.username}</div>
         </div>
     {/each}
 </div>

src/components/Radio.svelte

@@ -1,5 +1,5 @@
 <svelte:head>
-    <title>{`${ $current ? `${$current.metadata.title} ∴ ` : ''}Eldritch Radio`}</title>
+    <title>{`${ $current ? `${$current.title} ∴ ` : ''}Eldritch Radio`}</title>
 </svelte:head>
 
 <div class="app container">
@@ -28,8 +28,8 @@
     import Controls from '/components/Controls.svelte'
     import Queue from '/components/Queue.svelte'
     import Viewer from '/components/Viewer.svelte'
-    import { hashtagIterator, combinedIterator } from '/services/mastodon.js'
+    import { hashtagsIterator } from '/services/mastodon.js'
-    import { mkTracksIterator } from '/services/misc.js'
+    import { tracksIterator } from '/services/misc.js'
 
     import { domain, hashtags, queue, next, current, enqueueing, select } from '/store.js'
 
@@ -37,13 +37,10 @@
     let currentUnsubcribe = null
 
     onMount(async () => {
-        // const iterator = mkTracksIterator(hashtagIterator(get(domain), get(hashtags)[0]))
         const domainValue = get(domain)
         const hashtagsValue = get(hashtags)
 
-        const iterator = combinedIterator(
+        const iterator = tracksIterator(hashtagsIterator(domainValue, hashtagsValue))
-            hashtagsValue.map(hashtag => mkTracksIterator(hashtagIterator(domainValue, hashtag)))
-        )
 
         const { value: first } = await iterator.next()
 

src/components/Viewer.svelte

@@ -1,7 +1,7 @@
 <div class="playerBig">
     <div class="playerBig__player">
         <YoutubePlayer
-            id={$current ? $current.data.id : null}
+            id={$current ? $current.media.credentials.id : null}
             class="playerBig__iframe"
             paused={$paused}
             muted={$muted}
@@ -38,7 +38,6 @@
     import { secondsToElapsedTime } from '/services/misc.js'
     import { paused, muted, volume, current, selectNext, loading } from '/store.js'
 
-
     let ready = null
     let ended = null
     let error = null

src/services/mastodon.js

@@ -1,10 +1,8 @@
 import Observable from 'core-js-pure/features/observable'
-import { observableToAsyncIterator } from '/services/misc.js'
+import { observableToAsyncIterator, raceIterator } from '/services/misc.js'
 
 const LINK_RE = /<(.+?)>; rel="(\w+)"/gi
 
-export const fetchStatus = (domain, id) => fetch(`https://${domain}/api/v1/statuses/${id}`).then(x => x.json())
-
 function parseLinkHeader(link) {
     const links = {}
 
@@ -15,6 +13,9 @@ function parseLinkHeader(link) {
     return links
 }
 
+export const fetchStatus = (domain, id) => fetch(`https://${domain}/api/v1/statuses/${id}`).then(x => x.json())
+
+// Observable<{ domain : string, hashtag : string, status : Status}>
 export const hashtagStreamingObservable = (domain, hashtag) => {
     return new Observable(observer => {
         const onOpen = () => {
@@ -24,7 +25,7 @@ export const hashtagStreamingObservable = (domain, hashtag) => {
         const onStatus = event => {
             const status = JSON.parse(event.data)
             console.log(`Streaming ${domain} #${hashtag} : status ${status.id}`)
-            observer.next(status)
+            observer.next(processStatus(domain, status))
         }
 
         const onError = error => {
@@ -60,29 +61,19 @@ export async function* hashtagTimelineIterator (domain, hashtag) {
 
         console.log(`Timeline ${domain} #${hashtag} : fetched ${statuses.length} statuses`)
 
-        yield* statuses
+        yield* statuses.map(status => processStatus(domain, status))
     }
 }
 
-export async function* hashtagIterator(domain, hashtag) {
+export const hashtagIterator = (domain, hashtag) => {
-    const newerIterator = observableToAsyncIterator(hashtagStreamingObservable(domain, hashtag))
+    return raceIterator([
-    const olderIterator = hashtagTimelineIterator(domain, hashtag)
+        observableToAsyncIterator(hashtagStreamingObservable(domain, hashtag)),
-
+        hashtagTimelineIterator(domain, hashtag)
-    const iterators = [newerIterator, olderIterator]
+    ])
-    const values = iterators.map(iterator => iterator.next())
-
-    while (true) {
-        const promises = values.map((promise, index) => promise.then(result => ({ index, result })))
-        const { index, result: { done, value } } = await Promise.race(promises)
-
-        values[index] = iterators[index].next()
-
-        console.log(`Resolver ${domain} #${hashtag} : resolved with iterator ${index}`)
-        yield value
-    }
 }
 
-export async function* combinedIterator(iterators) {
+export async function* hashtagsIterator (domain, hashtags) {
+    const iterators = hashtags.map(hashtag => hashtagIterator(domain, hashtag))
     const values = iterators.map(iterator => iterator.next())
 
     while (true) {
@@ -91,14 +82,24 @@ export async function* combinedIterator(iterators) {
 
         const sorted = promisesValues
             .sort((a, b) =>{
-                new Date(a.result.value.status.created_at) - new Date(b.result.value.status.created_at)
+                a.result.value.date - b.result.value.date
             })
 
         const { index, result: { done, value } } = sorted[0]
-
         values[index] = iterators[index].next()
-
-        console.log(`CombinedResolver : resolved with iterator ${index}`)
         yield value
     }
 }
+
+const processStatus = (domain, status) => ({
+    title: null,
+    username: status.account.username,
+    date: new Date(status.createdAt),
+    content: status.content,
+    referer: {
+        url: status.url,
+        credentials: { type: 'mastodon', domain, id: status.id }
+    },
+    media: null
+})
+

src/services/misc.js

@@ -1,5 +1,5 @@
 import getUrls from 'get-urls'
-import { execPipe, asyncFilter, asyncMap } from 'iter-tools'
+import { execPipe, asyncFilter, asyncMap, map, findOr } from 'iter-tools'
 
 export const tap = f => x => {
     f(x)
@@ -80,65 +80,159 @@ export const secondsToElapsedTime = (seconds) => {
         .join(':')
 }
 
-export async function* mkTracksIterator(statusesIterator) {
+export async function* raceIterator(iterators) {
-    const knownStatus = new Set()
+    const values = iterators.map(iterator => iterator.next())
-    const knownYoutube = new Set()
 
-    const tracks = execPipe(
+    while (true) {
+        const promises = values.map((promise, index) => promise.then(result => ({ index, result })))
+        const { index, result: { done, value } } = await Promise.race(promises)
+
+        values[index] = iterators[index].next()
+        yield value
+    }
+}
+
+const mkMapSet = () => ({ set: new Set(), children: new Map() })
+
+const pathSet = () => {
+    const root = mkMapSet()
+
+    const has = (keys, value) => {
+        let x = root
+
+        for (const key of keys) {
+            if (x.children.has(key)) {
+                x = x.children.get(key)
+            } else {
+                return false
+            }
+        }
+
+        return x.set.has(value)
+    }
+
+    const add = (keys, value) => {
+        let x = root
+
+        for (const key of keys) {
+            if (!x.children.has(key)) {
+                x.children.set(key, mkMapSet())
+            }
+
+            x = x.children.get(key)
+        }
+
+        x.set.add(value)
+    }
+
+    return { root, has, add }
+}
+
+export async function* tracksIterator(statusesIterator) {
+    const known = pathSet()
+
+    yield* execPipe(
         statusesIterator,
-        asyncFilter(status => {
+        asyncFilter(knownByReferer(known)),
-            if (!status) {
+        asyncMap(processReferer),
+        asyncFilter(knownByMedia(known)),
+        asyncMap(processMedia)
+    )
+}
+
+const knownByReferer = known => track => {
+    if (!track) {
         console.error(`No status, should not happen here`)
         return false
     } else {
-                if (knownStatus.has(status.id)) {
+        switch (track.referer.credentials.type) {
-                    console.log(`Drop already processed status ${status.id}`)
+            default:
+                throw new Error()
+
+            case 'mastodon':
+                const path = [
+                    'referer',
+                    'mastodon',
+                    track.referer.credentials.domain
+                ]
+
+                const id = track.referer.credentials.id
+
+                if (known.has(path, id)) {
+                    console.log(`Drop already processed referer ${id}`)
                     return false
                 } else {
-                    knownStatus.add(status.id)
+                    known.add(path, id)
                     return true
                 }
         }
-        }),
+    }
-        asyncMap(status => ({ status, data: mkData(status) })),
+}
-        asyncFilter(({ status, data }) => {
+
-            if (!data) {
+const knownByMedia = known => track => {
-                console.log(`Drop non processable status ${status.id}`)
+    if (track !== null) {
+        switch (track.media.credentials.type) {
+            default:
+                throw new Error()
+
+            case 'youtube':
+                const path = [
+                    'media',
+                    'youtube'
+                ]
+
+                const id = track.media.credentials.id
+
+                if (known.has(path, id)) {
+                    console.log(`Drop already processed media ${id}`)
                     return false
                 } else {
-                if (knownYoutube.has(data.id)) {
+                    known.add(path, id)
-                    console.log(`Drop already processed youtube ${data.id}`)
-                    return false
-                } else {
-                    knownYoutube.add(data.id)
                     return true
                 }
         }
-        }),
+    } else {
-        asyncMap(async ({ status, data }) => ({ status, data, metadata: await mkMetadata(data) }))
+        return false
+    }
+}
+
+const processReferer = track => {
+    const urls = getUrls(track.content)
+
+    const media = execPipe(
+        urls,
+        map(parseSource),
+        findOr(null, x => x !== null)
     )
 
-    yield* tracks
+    if (media) {
-}
+        return { ...track, media }
-
+    } else {
-function mkData(status)
-{
-    const urls = getUrls(status.content)
-
-    for (const urlAsString of urls) {
-        const url = new URL(urlAsString)
-
-        if (['youtube.com', 'm.youtube.com', 'music.youtube.com'].includes(url.hostname) && url.searchParams.has('v')) {
-            return { url: urlAsString, id: url.searchParams.get('v') }
-        } else if (url.hostname === 'youtu.be') {
-            return { url: urlAsString, id: url.pathname.substring(1) }
-        }
-    }
-
         return null
     }
+}
 
-async function mkMetadata(entry) {
+const processMedia = async track => {
-    return fetch(`https://noembed.com/embed?url=https://www.youtube.com/watch?v=${entry.id}`)
+    const metadata = await fetchMetadata(track.media)
+    return { ...track, title: metadata.title }
+}
+
+const parseSource = (url) => {
+    const { hostname, pathname, searchParams } = new URL(url)
+
+    if (['youtube.com', 'm.youtube.com', 'music.youtube.com'].includes(hostname) && searchParams.has('v')) {
+        return { url, credentials: { type: 'youtube', id: searchParams.get('v') } }
+    } else if (hostname === 'youtu.be') {
+        return { url, credentials: { type: 'youtube', id: pathname.substring(1) } }
+    } else {
+        return null
+    }
+}
+
+const fetchMetadata = (media) => {
+    switch (media.credentials.type) {
+        case 'youtube':
+            return fetch(`https://noembed.com/embed?url=https://www.youtube.com/watch?v=${media.credentials.id}`)
                 .then(response => response.json())
     }
+}

src/store.js

@@ -31,7 +31,7 @@ export const canNext = derived([queue, index], ([$queue, $index]) => $index !==
 
 
 export const select = track => {
-    console.log(`Select ${track.metadata.title}`)
+    console.log(`Select ${track.title}`)
     current.set(track)
 }