Merge pull request #1 from QPixel/refactor/main

Refactor main so it works now

backend/.idea/workspace.xml

@@ -5,16 +5,7 @@
   </component>
   <component name="ChangeListManager">
     <list default="true" id="8a64704d-5500-41a6-aa4c-e275933fc58c" name="Changes" comment="">
-      <change afterPath="$PROJECT_DIR$/cmd/watchtogether/build.sh" afterDir="false" />
-      <change afterPath="$PROJECT_DIR$/../frontend/src/interfaces/SocketEvents.ts" afterDir="false" />
       <change beforePath="$PROJECT_DIR$/.idea/workspace.xml" beforeDir="false" afterPath="$PROJECT_DIR$/.idea/workspace.xml" afterDir="false" />
-      <change beforePath="$PROJECT_DIR$/cmd/watchtogether/main.go" beforeDir="false" afterPath="$PROJECT_DIR$/cmd/watchtogether/main.go" afterDir="false" />
-      <change beforePath="$PROJECT_DIR$/internal/ws/handlers.go" beforeDir="false" afterPath="$PROJECT_DIR$/internal/ws/handlers.go" afterDir="false" />
-      <change beforePath="$PROJECT_DIR$/internal/ws/hub.go" beforeDir="false" afterPath="$PROJECT_DIR$/internal/ws/hub.go" afterDir="false" />
-      <change beforePath="$PROJECT_DIR$/internal/ws/message.go" beforeDir="false" afterPath="$PROJECT_DIR$/internal/ws/message.go" afterDir="false" />
-      <change beforePath="$PROJECT_DIR$/../frontend/src/components/Container.tsx" beforeDir="false" afterPath="$PROJECT_DIR$/../frontend/src/components/Container.tsx" afterDir="false" />
-      <change beforePath="$PROJECT_DIR$/../frontend/src/components/Player.tsx" beforeDir="false" afterPath="$PROJECT_DIR$/../frontend/src/components/Player.tsx" afterDir="false" />
-      <change beforePath="$PROJECT_DIR$/../frontend/src/interfaces/Identity.ts" beforeDir="false" afterPath="$PROJECT_DIR$/../frontend/src/interfaces/Identity.ts" afterDir="false" />
       <change beforePath="$PROJECT_DIR$/../frontend/src/pages/player.tsx" beforeDir="false" afterPath="$PROJECT_DIR$/../frontend/src/pages/player.tsx" afterDir="false" />
     </list>
     <option name="SHOW_DIALOG" value="false" />

backend/internal/ws/handlers.go

@@ -12,14 +12,19 @@ func handleIdentifyEvent(message *Message) {
 	if id, ok := d["clientID"]; ok {
 		log.Infof("Client %s has sent identify event", id.(string))
 	}
+	user := d["user"].(map[string]interface{})
+	userId := user["id"].(string)
+	playhead := message.hub.State.playhead
+	paused := message.hub.State.paused
 	m := Message{
 		MessageData: MessageData{
 			Type: Identify,
 			Data: map[string]interface{}{
-				"admin":      true,
+				"admin":         message.hub.State.IsAdmin(userId),
 				"playlist":      "http://localhost:8081/BelleOpening.m3u8",
-				"controller": true,
+				"hasController": message.hub.State.IsController(userId),
-				"playhead":   0,
+				"playhead":      playhead,
+				"paused":        paused,
 				"user":          d["user"],
 			},
 		},
@@ -54,6 +59,14 @@ func handleSetPlayhead(message *Message) {
 		},
 	}
 	log.Infof("Received SetPlayhead event. playhead is at %s", d["playhead"])
+	err := message.hub.State.setPlayhead(d["playhead"].(float64))
+	if err != nil {
+		log.Errorf("unable to set playhead. %s", err)
+	}
+	err = message.hub.State.setPaused(d["paused"].(bool))
+	if err != nil {
+		log.Errorf("unable to set paused. %s", err)
+	}
 	for client := range message.Client.hub.Clients {
 		if client == message.Client {
 			continue

backend/internal/ws/hub.go

@@ -4,6 +4,9 @@ type Hub struct {
 	// Registered Clients
 	Clients map[*Client]bool
 
+	// State
+	State *State
+
 	// Inbound messages from the Clients
 	broadcast chan RawMessage
 
@@ -20,6 +23,7 @@ func NewHub() *Hub {
 		register:   make(chan *Client),
 		unregister: make(chan *Client),
 		Clients:    make(map[*Client]bool),
+		State:      NewState(),
 	}
 }
 

backend/internal/ws/state.go

@@ -0,0 +1,54 @@
+package ws
+
+import (
+	"errors"
+	"sync"
+)
+
+type State struct {
+	sync.RWMutex
+
+	playhead         float64
+	controllerUserId string
+	adminUserId      string
+	paused           bool
+}
+
+func NewState() *State {
+	return &State{
+		playhead:         0,
+		controllerUserId: "218072060923084802",
+		adminUserId:      "218072060923084802",
+	}
+}
+
+func (s *State) setPlayhead(playhead float64) error {
+	if s == nil {
+		return errors.New("unable to find state")
+	}
+	s.Lock()
+	defer s.Unlock()
+	s.playhead = playhead
+	return nil
+}
+func (s *State) setPaused(paused bool) error {
+	if s == nil {
+		return errors.New("unable to find state")
+	}
+	s.Lock()
+	defer s.Unlock()
+	s.paused = paused
+	return nil
+}
+func (s *State) IsController(userID string) bool {
+	if userID == s.controllerUserId {
+		return true
+	}
+	return false
+}
+func (s *State) IsAdmin(userID string) bool {
+	if userID == s.adminUserId {
+		return true
+	}
+	return false
+}

frontend/src/components/Player.tsx

@@ -1,90 +1,83 @@
-import { Box, css } from "@chakra-ui/react";
+import { Box } from "@chakra-ui/react";
-import React, { FC, useRef, useState } from "react";
+import React, { forwardRef } from "react";
 import ReactPlayer, { Config, ReactPlayerProps } from "react-player";
-import { MessageTypes } from "../interfaces/IMessage";
-import Message from "../util/Message";
-import MessageUtil from "../util/MessageUtil";
-import PlayerSocket from "../ws/websocket";
 
-type PlayerProps = {
+const Player = forwardRef<ReactPlayer, ReactPlayerProps>((props, ref) => {
-  id: string;
-  socket: PlayerSocket;
-} & ReactPlayerProps;
-
-interface ProgressProps {
-  played: number;
-  playedSeconds: number;
-  loaded: number;
-  loadedSeconds: number;
-}
-
-const Player: FC<PlayerProps> = (props) => {
-  const playerRef = useRef<ReactPlayer>(null);
-  const [paused, setPaused] = useState<boolean>(false);
-  const { socket } = props;
   const config: Config = {
     file: {
       forceHLS: true,
     },
   };
-  const onProgress = (state: ProgressProps) => {
+  // useEffect(() => {
-    if (paused) {
+  //   if (playerRef.current && typeof props.identity !== "undefined") {
-      socket?.send(
+  //     console.log(props.identity.playhead);
-        MessageUtil.encode(
+  //     playerRef.current.seekTo(props.identity.playhead);
-          new Message(MessageTypes.SetPlayhead, {
+  //     setPaused(props.identity.paused);
-            playhead: state.playedSeconds,
+  //   }
-            paused: true,
+  // }, []);
-          })
+  // socket.emitter.once(SocketEvents.SetPlayhead, (e) => {
-        )
+  //   console.log(e);
-      );
+  //   playerRef.current.seekTo(e.playhead);
-    }
+  //   setPaused(e.paused);
-    socket?.send(
+  // });
-      MessageUtil.encode(
+  // const onSeek = (playedSeconds: number) => {
-        new Message(MessageTypes.SetPlayhead, {
+  //   if (!props.identity.admin) return;
-          playhead: state.playedSeconds,
+  //   if (paused) {
-          paused: false,
+  //     socket?.send(
-        })
+  //       MessageUtil.encode(
-      )
+  //         new Message(MessageTypes.SetPlayhead, {
-    );
+  //           playhead: playedSeconds,
-  };
+  //           paused: true,
-  const onPause = () => {
+  //         })
-    setPaused(true);
+  //       )
-    socket?.send(
+  //     );
-      MessageUtil.encode(
+  //     return;
-        new Message(MessageTypes.SetPlayhead, {
+  //   }
-          playhead: playerRef.current.getCurrentTime(),
+  //   socket?.send(
-          paused: true,
+  //     MessageUtil.encode(
-        })
+  //       new Message(MessageTypes.SetPlayhead, {
-      )
+  //         playhead: playedSeconds,
-    );
+  //         paused: false,
-  };
+  //       })
-  const onPlay = () => {
+  //     )
-    setPaused(false);
+  //   );
-    socket?.send(
+  // };
-      MessageUtil.encode(
+  // const onPause = () => {
-        new Message(MessageTypes.SetPlayhead, {
+  //   if (!props.identity.admin) return;
-          playhead: playerRef.current.getCurrentTime(),
+  //   setPaused(true);
-          paused: false,
+  //   socket?.send(
-        })
+  //     MessageUtil.encode(
-      )
+  //       new Message(MessageTypes.SetPlayhead, {
-    );
+  //         playhead: playerRef.current.getCurrentTime(),
-  };
+  //         paused: true,
+  //       })
+  //     )
+  //   );
+  // };
+  // const onPlay = () => {
+  //   if (!props.identity.admin) return;
+  //   setPaused(false);
+  //   socket?.send(
+  //     MessageUtil.encode(
+  //       new Message(MessageTypes.SetPlayhead, {
+  //         playhead: playerRef.current.getCurrentTime(),
+  //         paused: false,
+  //       })
+  //     )
+  //   );
+  // };
   return (
     <Box height="100vh" width="100vw">
       <ReactPlayer
-        url={props.id}
+        url={props.url}
         width="100%"
         height="100%"
         config={config}
-        controls
+        ref={ref}
-        onPlay={onPlay}
-        onPause={onPause}
-        onProgress={onProgress}
-        ref={playerRef}
         {...props}
       />
     </Box>
   );
-};
+});
 
 export default Player;

frontend/src/hooks/useWS.ts

@@ -7,17 +7,22 @@ interface useWSProps {
 }
 
 // todo write websocket reconnector
-const useWS = ({ user }: useWSProps) => {
+const useWS = ({ user }: useWSProps): PlayerSocket | null => {
   if (typeof window === "undefined") {
     return;
   }
   // todo checkout usecallback
-  const [socket, setSocket] = useState<PlayerSocket>();
+  const [socket, setSocket] = useState<PlayerSocket>(null);
   useEffect(() => {
+    if (socket !== null) {
+      return;
+    }
     let internalSocket = new PlayerSocket(user);
     setSocket(internalSocket);
     return () => {
+      if (internalSocket.readyState !== WebSocket.OPEN) {
         return internalSocket.close();
+      }
     };
   }, []);
 

frontend/src/interfaces/Identity.ts

@@ -2,10 +2,11 @@ import IUser from "./IUser";
 
 interface IdentityData {
   admin?: boolean;
-  controller?: boolean;
+  hasController?: boolean;
   clientID?: string;
   playlist?: string;
-  playHead?: number;
+  playhead?: number;
+  paused?: boolean;
   user: IUser;
 }
 

frontend/src/interfaces/Playhead.ts

@@ -0,0 +1,4 @@
+export default interface SetPlayheadEvent {
+  playhead: number;
+  paused: boolean;
+}

frontend/src/interfaces/SocketEvents.ts

@@ -1,5 +1,7 @@
 enum SocketEvents {
   Identify = "Identify",
+  SetPlayhead = "SetPlayhead",
+  GetPlayhead = "GetPlayhead",
 }
 
 export default SocketEvents;

frontend/src/pages/index.tsx

@@ -8,7 +8,7 @@ import { Container } from "../components/Container";
 import { Footer } from "../components/Footer";
 import { Hero } from "../components/Hero";
 import { Main } from "../components/Main";
-import isDev from "../util/isDev";
+import { isDev } from "../util";
 
 const Index: NextPage = () => {
   return (

frontend/src/pages/player.tsx

@@ -1,47 +1,98 @@
-import { Socket } from "dgram";
 import { GetServerSideProps, NextPage } from "next";
 import { User } from "next-auth";
 import { getSession } from "next-auth/react";
-import dynamic from "next/dynamic";
 import Head from "next/head";
-import React, { useRef, useState } from "react";
+import React, { useEffect, useRef, useState } from "react";
 import ReactPlayer from "react-player";
-import { BaseReactPlayerProps } from "react-player/base";
 import { Container } from "../components/Container";
+import Player from "../components/Player";
 import useWS from "../hooks/useWS";
 import IdentityData from "../interfaces/Identity";
 import { MessageTypes } from "../interfaces/IMessage";
+import SetPlayheadEvent from "../interfaces/Playhead";
 import SocketEvents from "../interfaces/SocketEvents";
-import isBrowser from "../util/isBrowser";
+import { isBrowser } from "../util";
 import Message from "../util/Message";
 import MessageUtil from "../util/MessageUtil";
 
-const Player = dynamic(() => import("../components/Player"), { ssr: false });
-
 interface PlayerPageProps {
   user: User;
 }
 // types for the function
 
 const PlayerPage: NextPage<PlayerPageProps> = ({ user }) => {
-  // const playerRef = useRef<ReactPlayer>();
   const socket = useWS({ user });
+  const playerRef = useRef<ReactPlayer>();
   const [id, setID] = useState<string>("");
   const [identity, setIdentity] = useState<IdentityData>();
+  const [paused, setPaused] = useState<boolean>(true);
+
+  useEffect(() => {
     if (isBrowser() && typeof socket !== "undefined") {
-    socket.emitter.on(SocketEvents.Identify, (e: IdentityData) => {
+      socket?.emitter.once(SocketEvents.Identify, (e: IdentityData) => {
-      console.log(e);
         setID(e.playlist);
         setIdentity(e);
+        playerRef?.current.seekTo(e.playhead);
+        setPaused(e.paused);
+      });
+      socket?.emitter.on(SocketEvents.SetPlayhead, (e: SetPlayheadEvent) => {
+        console.log(e.paused);
+        setPaused(e.paused);
+        playerRef.current.seekTo(e.playhead);
       });
     }
+  }, [socket]);
+  const onPlay = () => {
+    if (!identity.admin) return;
+    setPaused(false);
+    socket?.send(
+      MessageUtil.encode(
+        new Message(MessageTypes.SetPlayhead, {
+          playhead: playerRef.current.getCurrentTime(),
+          paused: false,
+        })
+      )
+    );
+  };
+  const onSeek = (playedSeconds: number) => {
+    if (!identity.admin) return;
+    socket.send(
+      MessageUtil.encode(
+        new Message(MessageTypes.SetPlayhead, {
+          playhead: playedSeconds,
+          paused: paused,
+        })
+      )
+    );
+  };
+  const onPause = () => {
+    console.log("running now");
+    if (!identity.admin) return;
+    setPaused(true);
+    socket?.send(
+      MessageUtil.encode(
+        new Message(MessageTypes.SetPlayhead, {
+          playhead: playerRef.current.getCurrentTime(),
+          paused: true,
+        })
+      )
+    );
+  };
   return (
     <>
       <Head>
         <title>Watch Together</title>
       </Head>
       <Container height="100vh" background={"#000"}>
-        <Player id={id} socket={socket} />
+        <Player
+          url={id}
+          onPlay={onPlay}
+          onPause={onPause}
+          onSeek={onSeek}
+          controls={identity?.hasController}
+          playing={!paused}
+          ref={playerRef}
+        />
       </Container>
     </>
   );

frontend/src/util/Handler.ts

@@ -1 +0,0 @@
-export default class Handler {}

frontend/src/util/api.tsx

@@ -1,3 +0,0 @@
-const useAPI = () => {};
-
-export default useAPI;

frontend/src/util/index.ts

@@ -0,0 +1,6 @@
+export function isDev() {
+  return process.env.NODE_ENV === "development";
+}
+export function isBrowser() {
+  return typeof window !== "undefined";
+}

frontend/src/util/isBrowser.ts

@@ -1,3 +0,0 @@
-export default function isBrowser() {
-  return typeof window !== "undefined";
-}

frontend/src/util/isDev.ts

@@ -1,3 +0,0 @@
-export default function isDev() {
-  return process.env.NODE_ENV === "development";
-}

frontend/src/ws/websocket.ts

@@ -25,6 +25,10 @@ export default class PlayerSocket extends Websocket {
     this.onmessage = this.onMessage;
     this.onclose = this.onClose;
   }
+  close(code?: number, reason?: string): void {
+    this.emitter.removeAllListeners();
+    super.close(code, reason);
+  }
   onMessage(evt: MessageEvent<any>) {
     let message = MessageUtil.decode(evt.data);
     if (message.type === MessageTypes["Ping"]) {
@@ -65,7 +69,6 @@ export default class PlayerSocket extends Websocket {
   }
   onClose(event: CloseEvent) {
     console.log("[WS] socket connection closed");
-    console.log(event);
     this.emitter.emit("closed");
   }
   get open() {