fix streaming gap issue

routes/_actions/addStatusOrNotification.js

@@ -76,8 +76,12 @@ const lazilyProcessFreshUpdates = throttle((instanceName, timelineName) => {
 }, 5000)
 
 export function addStatusOrNotification (instanceName, timelineName, newStatusOrNotification) {
+  addStatusesOrNotifications(instanceName, timelineName, [newStatusOrNotification])
+}
+
+export function addStatusesOrNotifications (instanceName, timelineName, newStatusesOrNotifications) {
   let freshUpdates = store.getForTimeline(instanceName, timelineName, 'freshUpdates') || []
-  freshUpdates.push(newStatusOrNotification)
+  freshUpdates = freshUpdates.concat(newStatusesOrNotifications)
   freshUpdates = uniqBy(freshUpdates, _ => _.id)
   store.setForTimeline(instanceName, timelineName, {freshUpdates: freshUpdates})
   lazilyProcessFreshUpdates(instanceName, timelineName)

routes/_store/observers/timelineObservers.js

@@ -1,15 +1,14 @@
 import { updateInstanceInfo } from '../../_actions/instances'
 import { createStream } from '../../_actions/streaming'
+import { getTimeline } from '../../_api/timelines'
+import { addStatusesOrNotifications } from '../../_actions/addStatusOrNotification'
 
 export function timelineObservers (store) {
   // stream to watch for local/federated/etc. updates. home and notification
   // updates are handled in timelineObservers.js
   let currentTimelineStream
 
-  store.observe('currentTimeline', async (currentTimeline) => {
-    if (!process.browser) {
-      return
-    }
+  function shutdownPreviousStream () {
     if (currentTimelineStream) {
       currentTimelineStream.close()
       currentTimelineStream = null
@@ -17,26 +16,61 @@ export function timelineObservers (store) {
         window.currentTimelineStream = null
       }
     }
-    if (!currentTimeline) {
+  }
+
+  function shouldObserveTimeline (timeline) {
+    return timeline &&
+      !(
+        timeline !== 'local' &&
+        timeline !== 'federated' &&
+        !timeline.startsWith('list/') &&
+        !timeline.startsWith('tag/')
+      )
+  }
+
+  store.observe('currentTimeline', async (currentTimeline) => {
+    if (!process.browser) {
       return
     }
-    if (currentTimeline !== 'local' &&
-        currentTimeline !== 'federated' &&
-        !currentTimeline.startsWith('list/') &&
-        !currentTimeline.startsWith('tag/')) {
+
+    shutdownPreviousStream()
+
+    if (!shouldObserveTimeline(currentTimeline)) {
       return
     }
 
     let currentInstance = store.get('currentInstance')
+    let accessToken = store.get('accessToken')
     await updateInstanceInfo(currentInstance)
-    let instanceInfo = store.get('currentInstanceInfo')
-    if (!(instanceInfo &&
-        store.get('currentInstance') === currentInstance &&
-        store.get('currentTimeline') === currentTimeline)) {
+
+    let checkInstanceAndTimelineAreUnchanged = () => (
+      store.get('currentInstance') === currentInstance &&
+      store.get('currentTimeline') === currentTimeline
+    )
+
+    if (!checkInstanceAndTimelineAreUnchanged()) {
       return
     }
 
-    let accessToken = store.get('accessToken')
+    let timelineItemIds = store.getForTimeline(currentInstance,
+      currentTimeline, 'timelineItemIds')
+    let firstTimelineItemId = timelineItemIds && timelineItemIds[0]
+
+    if (firstTimelineItemId) {
+      // fill in the "streaming gap" – i.e. fetch the most recent 20 items so that there isn't
+      // a big gap in the timeline if you haven't looked at it in awhile
+      // TODO: race condition here, should start streaming while this request is ongoing
+      let newTimelineItems = await getTimeline(currentInstance, accessToken, currentTimeline,
+        null, firstTimelineItemId)
+      if (newTimelineItems.length) {
+        addStatusesOrNotifications(currentInstance, currentTimeline, newTimelineItems)
+      }
+      if (!checkInstanceAndTimelineAreUnchanged()) {
+        return
+      }
+    }
+
+    let instanceInfo = store.get('currentInstanceInfo')
     currentTimelineStream = createStream(instanceInfo.urls.streaming_api,
       currentInstance, accessToken, currentTimeline)
 

tests/spec/107-streaming-gap.js

@@ -0,0 +1,29 @@
+import { foobarRole } from '../roles'
+import {
+  getNthStatus, homeNavButton, localTimelineNavButton, sleep
+} from '../utils'
+import {
+  postAs
+} from '../serverActions'
+
+fixture`107-streaming-gap.js`
+  .page`http://localhost:4002`
+
+test('fills in a status posted while away from timeline', async t => {
+  let timeout = 20000
+
+  await t.useRole(foobarRole)
+    .click(localTimelineNavButton)
+    .hover(getNthStatus(0))
+  await postAs('admin', 'heyo')
+  await t.expect(getNthStatus(0).innerText).contains('heyo', {timeout})
+    .click(homeNavButton)
+    .hover(getNthStatus(0))
+  await postAs('admin', 'posted this while you were away!')
+  await t.expect(getNthStatus(0).innerText).contains('posted this while you were away!', {timeout})
+    .click(localTimelineNavButton)
+    .expect(getNthStatus(0).innerText).contains('posted this while you were away!', {timeout})
+  await sleep(2000)
+  await postAs('admin', 'posted this while you were watching')
+  await t.expect(getNthStatus(0).innerText).contains('posted this while you were watching', {timeout})
+})

tests/utils.js

@@ -10,6 +10,7 @@ export const modalDialogContents = $('.modal-dialog-contents')
 export const closeDialogButton = $('.close-dialog-button')
 export const notificationsNavButton = $('nav a[href="/notifications"]')
 export const homeNavButton = $('nav a[href="/"]')
+export const localTimelineNavButton = $('nav a[href="/local"]')
 export const searchNavButton = $('nav a[href="/search"]')
 export const formError = $('.form-error-user-error')
 export const composeInput = $('.compose-box-input')