Fix comment behavior

apps/server/src/models/comment.ts

@@ -25,6 +25,22 @@ export const Comments = (client: PrismaClient['comment']) =>
         isDeleted: comment.isDeleted
       };
     },
+    getComments: async (liveId: number, lastCommentId = 0) => {
+      const comments = await client.findMany({
+        where: {
+          liveId,
+          id: {
+            gt: lastCommentId
+          }
+        },
+        orderBy: {
+          id: 'desc'
+        },
+        take: 100
+      });
+
+      return comments;
+    },
     createViaLocal: async (userId: number, liveId: number, content: string) => {
       const data = await client.create({
         data: {
@@ -39,7 +55,7 @@ export const Comments = (client: PrismaClient['comment']) =>
         getLiveUpdateKey(liveId),
         JSON.stringify({
           type: 'comment:created',
-          data: result
+          data: [result]
         } as LiveUpdateCommentCreated)
       );
 
@@ -70,7 +86,7 @@ export const Comments = (client: PrismaClient['comment']) =>
         getLiveUpdateKey(liveId),
         JSON.stringify({
           type: 'comment:created',
-          data: result
+          data: [result]
         } as LiveUpdateCommentCreated)
       );
 

apps/server/src/streaming/index.ts

@@ -1,11 +1,12 @@
 import { Server } from 'http';
 import WebSocket from 'ws';
 import { pathToRegexp } from 'path-to-regexp';
-import { lives } from '../models';
+import { comments, lives } from '../models';
 import { pubsub } from '../services/redis/pubsub/client';
 import { getLiveUpdateKey } from '../services/redis/pubsub/keys';
 import { UserToken } from '../services/redis/user-token';
 import { jwtCommentViewer } from '../services/jwt';
+import { LiveUpdateCommentCreated } from 'api-types/streaming/live-update';
 
 const userToken = new UserToken();
 
@@ -32,12 +33,11 @@ export class Streaming {
 
   private handleConnection(socket: WebSocket, Url: string) {
     const url = new URL(Url, process.env.SERVER_ENDPOINT);
-    const token = url.searchParams.get('token') || undefined;
 
     const live = liveUpdateRegexp.exec(url.pathname);
     if (live) {
       const liveId = Number(live[1]);
-      return this.handleV1LiveUpdate(socket, liveId, token);
+      return this.handleV1LiveUpdate(socket, liveId, url.searchParams);
     }
 
     return this.closeConnection(socket, 'invalid_path');
@@ -46,8 +46,10 @@ export class Streaming {
   private async handleV1LiveUpdate(
     socket: WebSocket,
     liveId: number,
-    token?: string
+    params: URLSearchParams
   ) {
+    const token = params.get('token');
+    const commentAfter = Number(params.get('commentAfter') || 0);
     const verifyAsStreamer =
       token && (await jwtCommentViewer.verifyToken(token, liveId));
     console.log(token, verifyAsStreamer);
@@ -71,6 +73,14 @@ export class Streaming {
 
     console.log('websocket connected', liveId);
 
+    const recentComments = await comments.getComments(liveId, commentAfter);
+    socket.send(
+      JSON.stringify({
+        type: 'comment:created',
+        data: recentComments.map(comments.getPublic)
+      } as LiveUpdateCommentCreated)
+    );
+
     const handle = {
       event: getLiveUpdateKey(liveId),
       callback: (message: string) => {

apps/web/pages/app/[locale]/tenants/[slug]/[id]/broadcast-via-browser.page.tsx

@@ -68,6 +68,7 @@ const Page: NextPage<PageProps<Props, Params & PathProps>> = ({
   const toast = useToast();
   const router = useRouter();
   const isInitializedRef = useRef(false);
+  const commentsRef = useRef<HTMLDivElement>(null);
   const [liveId] = useConvertLiveId(slug, id);
   const [stream, mutate] = useStream(liveId);
   const live = stream?.live;
@@ -186,6 +187,16 @@ const Page: NextPage<PageProps<Props, Params & PathProps>> = ({
     void mutate();
   }, [realtimeLive, mutate]);
 
+  useEffect(() => {
+    if (!commentsRef.current) {
+      return;
+    }
+
+    if (commentsRef.current.scrollTop > -300) {
+      commentsRef.current.scrollTop = commentsRef.current.scrollHeight;
+    }
+  }, [comments]);
+
   return (
     <Container padding={0}>
       <Head>
@@ -235,6 +246,7 @@ const Page: NextPage<PageProps<Props, Params & PathProps>> = ({
           py={2}
           align="stretch"
           flexDirection="column-reverse"
+          ref={commentsRef}
         >
           {!isConnectingStreaming && (
             <Box>

apps/web/utils/hooks/api/use-live-realtime.ts

@@ -1,6 +1,6 @@
 import { CommentPublic, LivePublic } from 'api-types/common/types';
 import { useCallback, useEffect, useReducer, useRef, useState } from 'react';
-import { client, wsURL } from '~/utils/api/client';
+import { wsURL } from '~/utils/api/client';
 import { useAuth } from '../use-auth';
 import { useAPIError } from './use-api-error';
 import {
@@ -23,18 +23,18 @@ const commentReducer = (
     return [];
   }
 
-  let comment: CommentPublic;
+  let data: CommentPublic[] = [];
   if ('type' in action) {
     if (action.type === 'comment:deleted') {
       return state.filter(comment => comment.id !== action.data.id);
     }
 
-    comment = action.data;
+    data = state.concat(action.data);
   } else {
-    comment = action;
+    data = state.concat(action);
   }
 
-  const data = state.concat(comment).sort((a, b) => b.id - a.id);
+  data.sort((a, b) => b.id - a.id);
   data.splice(100);
 
   const result = [];
@@ -50,10 +50,11 @@ const commentReducer = (
 };
 
 export const useLiveRealtime = (liveId?: number, viewerToken?: string) => {
-  const { token, headers } = useAuth();
+  const { token } = useAuth();
   const tokenRef = useRef(token);
   const socketRef = useRef<WebSocket>();
   const timeoutRef = useRef<NodeJS.Timeout>();
+  const lastCommentIdRef = useRef<number>(0);
   const needConnectingRef = useRef(true);
   const [isConnecting, setIsConnecting] = useState(false);
   const [comments, setComment] = useReducer(commentReducer, []);
@@ -85,7 +86,7 @@ export const useLiveRealtime = (liveId?: number, viewerToken?: string) => {
 
     try {
       const Token = viewerToken || tokenRef.current || '';
-      const url = `${wsURL}/websocket/v1/live/${liveId}?token=${Token}`;
+      const url = `${wsURL}/websocket/v1/live/${liveId}?token=${Token}&commentAfter=${lastCommentIdRef.current}`;
       const ws = new WebSocket(url);
       socketRef.current = ws;
       setIsConnecting(true);
@@ -152,43 +153,18 @@ export const useLiveRealtime = (liveId?: number, viewerToken?: string) => {
 
     return () => {
       needConnectingRef.current = false;
+      lastCommentIdRef.current = 0;
       setComment(undefined);
       setLive(undefined);
       disconnect();
     };
   }, [liveId, viewerToken, connect, disconnect]);
 
   useEffect(() => {
-    if (!liveId) {
-      return;
+    if (comments.length > 0) {
+      lastCommentIdRef.current = comments[0].id;
     }
-
-    void (async () => {
-      try {
-        if (viewerToken) {
-          const { body } = await client.v1.lives._liveId(liveId).comments.get({
-            query: {
-              viewerToken
-            }
-          });
-
-          body.forEach(comment => {
-            setComment(comment);
-          });
-        } else {
-          const { body } = await client.v1.lives._liveId(liveId).comments.get({
-            headers
-          });
-
-          body.forEach(comment => {
-            setComment(comment);
-          });
-        }
-      } catch (e) {
-        console.warn(e);
-      }
-    })();
-  }, [viewerToken, liveId, headers]);
+  }, [comments]);
 
   return { comments, live, isConnecting, reconnect } as const;
 };

packages/api-types/streaming/live-update.ts

@@ -7,7 +7,7 @@ export interface LiveUpdateMessageBase {
 
 export interface LiveUpdateCommentCreated extends LiveUpdateMessageBase {
   type: 'comment:created';
-  data: CommentPublic;
+  data: CommentPublic[];
 }
 
 export interface LiveUpdateCommentDeleted extends LiveUpdateMessageBase {