Add notifications endpoint with streaming websocket updates

Author: markerikson

src/features/notifications/notificationsSlice.js

@@ -2,9 +2,79 @@ import {
   createSlice,
   createAsyncThunk,
   createEntityAdapter,
+  createSelector,
 } from '@reduxjs/toolkit'
 
 import { client } from '../../api/client'
+import { forceGenerateNotifications } from '../../api/server'
+import { apiSlice } from '../api/apiSlice'
+
+export const extendedApi = apiSlice.injectEndpoints({
+  endpoints: (builder) => ({
+    getNotifications: builder.query({
+      query: () => '/notifications',
+      transformResponse: (res) => res.notifications,
+      async onCacheEntryAdded(
+        arg,
+        { updateCachedData, cacheDataLoaded, cacheEntryRemoved }
+      ) {
+        // create a websocket connection when the cache subscription starts
+        const ws = new WebSocket('ws://localhost')
+        try {
+          // wait for the initial query to resolve before proceeding
+          await cacheDataLoaded
+
+          // when data is received from the socket connection to the server,
+          // update our query result with the received message
+          const listener = (event) => {
+            const message = JSON.parse(event.data)
+            switch (message.type) {
+              case 'notifications': {
+                updateCachedData((draft) => {
+                  // Insert all received notifications from the websocket
+                  // into the existing RTKQ cache array
+                  draft.push(...message.payload)
+                  draft.sort((a, b) => b.date.localeCompare(a.date))
+                })
+                break
+              }
+              default:
+                break
+            }
+          }
+
+          ws.addEventListener('message', listener)
+        } catch {
+          // no-op in case `cacheEntryRemoved` resolves before `cacheDataLoaded`,
+          // in which case `cacheDataLoaded` will throw
+        }
+        // cacheEntryRemoved will resolve when the cache subscription is no longer active
+        await cacheEntryRemoved
+        // perform cleanup steps once the `cacheEntryRemoved` promise resolves
+        ws.close()
+      },
+    }),
+  }),
+})
+
+export const { useGetNotificationsQuery } = extendedApi
+
+const emptyNotifications = []
+
+export const selectNotificationsResult = extendedApi.endpoints.getNotifications.select()
+
+const selectNotificationsData = createSelector(
+  selectNotificationsResult,
+  (notificationsResult) => notificationsResult.data ?? emptyNotifications
+)
+
+export const fetchNotificationsWebsocket = () => (dispatch, getState) => {
+  const allNotifications = selectNotificationsData(getState())
+  const [latestNotification] = allNotifications
+  const latestTimestamp = latestNotification?.date ?? ''
+  // Hardcode a call to the mock server to simulate a server push scenario over websockets
+  forceGenerateNotifications(latestTimestamp)
+}
 
 const notificationsAdapter = createEntityAdapter({
   sortComparer: (a, b) => b.date.localeCompare(a.date),