From b4bb4004f128c22ec1eef6884ebbf5a11e5ead49 Mon Sep 17 00:00:00 2001
From: Inveracity <cbaklid@gmail.com>
Date: Fri, 21 Jul 2023 20:18:58 +0200
Subject: [PATCH 1/2] move logout to the top right

---
 frontend/src/components/Login.svelte  |  3 +++
 frontend/src/components/Logout.svelte |  5 +++--
 frontend/src/components/Users.svelte  | 17 +++++++++++++++++
 frontend/src/routes/+page.svelte      |  3 ++-
 4 files changed, 25 insertions(+), 3 deletions(-)
 create mode 100644 frontend/src/components/Users.svelte

diff --git a/frontend/src/components/Login.svelte b/frontend/src/components/Login.svelte
index b62f004..a0eb48a 100644
--- a/frontend/src/components/Login.svelte
+++ b/frontend/src/components/Login.svelte
@@ -3,6 +3,7 @@
 	import { currentUser, pb } from '$lib/pocketbase';
 	import { Connect } from '$lib/grpc';
 	import Toast from './Toast.svelte';
+	import Logout from './Logout.svelte';
 	let toast: Toast;
 	let password = '';
 	let username = '';
@@ -35,5 +36,7 @@
 			<input class="input input-accent" type="text" bind:value={$server} placeholder="myserver" />
 			<button class="btn btn-secondary" on:click={login}> Login </button>
 		</form>
+	{:else}
+		<Logout />
 	{/if}
 </div>
diff --git a/frontend/src/components/Logout.svelte b/frontend/src/components/Logout.svelte
index 448c89c..6ed4c27 100644
--- a/frontend/src/components/Logout.svelte
+++ b/frontend/src/components/Logout.svelte
@@ -9,9 +9,10 @@
 	}
 </script>
 
-<div class="bg-neutral w-40">
+<div>
 	<ul class="menu menu-horizontal gap-3">
-		<button class="flex-1 btn btn-ghost" on:click={logout}>
+		<button class="flex gap-2" on:click={logout}>
+			<p>sign out</p>
 			<div class="h-5 w-5">
 				<ArrowRightFromBracketSolid />
 			</div>
diff --git a/frontend/src/components/Users.svelte b/frontend/src/components/Users.svelte
new file mode 100644
index 0000000..76badb8
--- /dev/null
+++ b/frontend/src/components/Users.svelte
@@ -0,0 +1,17 @@
+<script>
+	import { users } from '$lib/stores/users';
+</script>
+
+<div class="flex w-40 bg-neutral-focus overflow-auto">
+	<div class="flex flex-col w-full">
+		{#each $users as user}
+			<div class="flex justify-center m-1 p-1">
+				<div class="avatar {user.presence ? 'online' : 'offline'} justify-center">
+					<div class="flex w-16 ring ring-neutral ring-offset-base-100 ring-offset-2">
+						{user.name}
+					</div>
+				</div>
+			</div>
+		{/each}
+	</div>
+</div>
diff --git a/frontend/src/routes/+page.svelte b/frontend/src/routes/+page.svelte
index 4dbeb62..17ff330 100644
--- a/frontend/src/routes/+page.svelte
+++ b/frontend/src/routes/+page.svelte
@@ -9,6 +9,7 @@
 	import Navbar from '../components/Navbar.svelte';
 	import Logout from '../components/Logout.svelte';
 	import Loading from '../components/Loading.svelte';
+	import Users from '../components/Users.svelte';
 </script>
 
 <div class="flex flex-col h-screen w-screen">
@@ -20,12 +21,12 @@
 		<Loading />
 	{:else if $status === 'connected'}
 		<div class="flex flex-row h-full w-full overflow-y-auto">
+			<Users />
 			<History />
 			<Channels />
 		</div>
 		<div class="flex flex-row w-full">
 			<Input />
-			<Logout />
 		</div>
 	{/if}
 </div>

From 80ea937efd5bf682cca9f5bbda7df46716483527 Mon Sep 17 00:00:00 2001
From: Inveracity <cbaklid@gmail.com>
Date: Fri, 21 Jul 2023 20:19:30 +0200
Subject: [PATCH 2/2] show user presence

---
 frontend/src/lib/grpc.ts         | 19 +++++++--
 frontend/src/lib/pocketbase.ts   |  9 +++--
 frontend/src/lib/stores/users.ts | 22 +++++++++--
 relay/internal/server/server.go  | 67 ++++++++++++++++++--------------
 4 files changed, 77 insertions(+), 40 deletions(-)

diff --git a/frontend/src/lib/grpc.ts b/frontend/src/lib/grpc.ts
index 9c0ddcc..93ec787 100644
--- a/frontend/src/lib/grpc.ts
+++ b/frontend/src/lib/grpc.ts
@@ -13,6 +13,7 @@ import type { ChatMessage } from '$lib/proto/chat/v1/chat';
 import type { Message, OutgoingMessage } from './types';
 import { currentUser, pb } from './pocketbase';
 import { channels } from './stores/channel';
+import { users, type User } from './stores/users';
 
 export const chat_cache = persisted(
   'chatmessages', // storage
@@ -92,6 +93,7 @@ export const Disconnect = async () => {
   chat_cache.set({ lastTs: "0" })
   messages.reset();
   controller.abort();
+  users.upd({ name: pb.authStore.model?.name, presence: false });
 };
 
 export const SendMessage = (msg: OutgoingMessage) => {
@@ -129,7 +131,7 @@ const filtered = (msg: ChatMessage, lastTs: string): boolean => {
     return true;
   }
 
-  if (msg.channelId === "system" && msg.userId !== pb.authStore.model?.name) {
+  if (msg.channelId === "system") {
     return filter_system_messages(msg);
   }
 
@@ -152,13 +154,24 @@ const timestampToDate = (timestamp: string): string => {
 }
 
 const filter_system_messages = (msg: ChatMessage): boolean => {
-
   // Tell UI to show new channel when another user adds one
-  if (msg.text.startsWith("channel_add")) {
+  if (msg.text.startsWith("channel_add") && msg.userId !== pb.authStore.model?.name) {
     const channel_name = msg.text.split(" ")[1]
     console.log(channel_name)
     channels.add(channel_name);
   }
 
+  if (msg.text.startsWith("connected")) {
+    console.log("connected", msg.userId)
+    const user: User = { name: msg.userId, presence: true }
+    users.upd(user);
+  }
+
+  if (msg.text.startsWith("disconnected")) {
+    console.log("disconnected", msg.userId)
+    const user: User = { name: msg.userId, presence: false }
+    users.upd(user);
+  }
+
   return true;
 }
diff --git a/frontend/src/lib/pocketbase.ts b/frontend/src/lib/pocketbase.ts
index 9a15cec..0b2408e 100644
--- a/frontend/src/lib/pocketbase.ts
+++ b/frontend/src/lib/pocketbase.ts
@@ -2,7 +2,7 @@ import { env } from '$env/dynamic/public';
 import PocketBase from 'pocketbase';
 import { writable } from 'svelte/store';
 import { channels } from './stores/channel';
-import { users } from './stores/users';
+import { users, type User } from './stores/users';
 
 export const pb = new PocketBase(env.PUBLIC_POCKETBASE_URL);
 export const currentUser = writable(pb.authStore.model);
@@ -29,12 +29,13 @@ export const writeChannel = async (channelName: string) => {
 }
 
 export const fetchUsers = async () => {
-  const records = await pb.collection('users').getFullList({
+  const records: User[] = await pb.collection('users').getFullList({
     sort: 'created',
   });
 
   // convert records to array and set in channels store
-  users.set(records.map((record) => {
-    return record.name;
+  users.set(records.map((record: User) => {
+    const user: User = { name: record.name, presence: false };
+    return user;
   }));
 }
diff --git a/frontend/src/lib/stores/users.ts b/frontend/src/lib/stores/users.ts
index 4daed8e..89c5764 100644
--- a/frontend/src/lib/stores/users.ts
+++ b/frontend/src/lib/stores/users.ts
@@ -1,12 +1,26 @@
 import { writable } from 'svelte/store';
 
+export interface User {
+  name: string;
+  presence: boolean;
+}
+
 function createUserList() {
-  const { subscribe, update } = writable<string[]>([]);
+  const { subscribe, update } = writable<User[]>([]);
   return {
     subscribe,
-    add: (user: string) => update(users => [...users, user]),
-    remove: (user: string) => update(users => users.filter(c => c !== user)),
-    set: (users: string[]) => update(_ => users),
+    add: (user: User) => update(users => [...users, user]),
+    remove: (user: User) => update(users => users.filter(c => c !== user)),
+    set: (users: User[]) => update(_ => users),
+    upd: (user: User) => update(users => {
+      const index = users.findIndex(u => u.name === user.name);
+      if (index === -1) {
+        console.log("user not found", user.name)
+        return users;
+      }
+      users[index] = user;
+      return users;
+    }),
   };
 }
 
diff --git a/relay/internal/server/server.go b/relay/internal/server/server.go
index 5541686..ae96847 100644
--- a/relay/internal/server/server.go
+++ b/relay/internal/server/server.go
@@ -18,11 +18,13 @@ import (
 
 type Server struct {
 	pb.UnimplementedChatServiceServer
-	cache   *cache.Cache
-	natsURL string
-	pbURL   string
-	pbAdmin string
-	pbPass  string
+	cache    *cache.Cache
+	queue    *queue.Queue
+	streamid string
+	natsURL  string
+	pbURL    string
+	pbAdmin  string
+	pbPass   string
 }
 
 func NewServer(natsURL, pbURL, pbAdmin, pbPass string, cache *cache.Cache) *Server {
@@ -37,65 +39,64 @@ func NewServer(natsURL, pbURL, pbAdmin, pbPass string, cache *cache.Cache) *Serv
 
 func (s *Server) Connect(in *pb.ConnectRequest, srv pb.ChatService_ConnectServer) error {
 	// Create a unique streamid for this connection
-	streamid := RandStringRunes(10)
+	s.streamid = RandStringRunes(10)
 
-	log.Printf("GRPC %s: user %s connected to server %s", streamid, in.UserId, in.ServerId)
+	log.Printf("GRPC %s: user %s connected to server %s", s.streamid, in.UserId, in.ServerId)
 
 	auth := auth.New(s.pbURL, s.pbAdmin, s.pbPass)
 
 	authed, err := auth.VerifyUserToken(in.Jwt)
 	if err != nil {
-		log.Printf("GRPC %s: error verifying jwt: %v", streamid, err)
+		log.Printf("GRPC %s: error verifying jwt: %v", s.streamid, err)
 		return fmt.Errorf("error verifying jwt")
 	}
 
 	if !authed {
-		log.Printf("GRPC %s: user %s not authorized", streamid, in.UserId)
+		log.Printf("GRPC %s: user %s not authorized", s.streamid, in.UserId)
 		return fmt.Errorf("user not authorized")
 	}
 
-	// Create a NATS queue subscriber for this streamid
-	queue := queue.NewQueue(s.natsURL, streamid)
+	// Create a NATS queue subscriber for this s.streamid
+	s.queue = queue.NewQueue(s.natsURL, s.streamid)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	go queue.Subscribe(ctx, in.ServerId)
+	go s.queue.Subscribe(ctx, in.ServerId)
 
 	// send a "connected" message to the client to tell the client it successfully connected
-	srv.Send(systemMessage("connected"))
-
+	srv.Send(systemMessage("connected", "server"))
 	// getPastMessages
 	if err := s.getPastMessages(srv, in); err != nil {
 		log.Printf("error getting past messages: %v", err)
 		return err
 	}
 
-	go ping(ctx, srv, cancel, streamid)
+	go s.ping(ctx, srv, in, cancel)
 	// Receive messages from the NATS loop and forward them to the client
 	for {
 		select {
 		case <-ctx.Done():
-			log.Printf("GRPC %s: %s disconnected from %s. Global context cancelled.", streamid, in.UserId, in.ServerId)
+			log.Printf("GRPC %s: %s disconnected from %s. Global context cancelled.", s.streamid, in.UserId, in.ServerId)
 			return nil
 
 		default:
 			if err := srv.Context().Err(); err != nil {
-				log.Printf("GRPC %s: Server found the context to be done in the default case, cancelling global context", streamid)
+				log.Printf("GRPC %s: Server found the context to be done in the default case, cancelling global context", s.streamid)
 				cancel()
 				return nil
 			}
 
-			for message := range *queue.Messages {
-				if err := relay(message, srv, cancel, streamid); err != nil {
-					queue.ErrCh <- err
+			for message := range *s.queue.Messages {
+				if err := relay(message, srv, cancel, s.streamid); err != nil {
+					s.queue.ErrCh <- err
 				}
 			}
 		}
 	}
 }
 
-// Send receives a message from the client and publishes it to the NATS server
+// Send: receives a message from the client and publishes it to the NATS server
 func (s *Server) Send(ctx context.Context, in *pb.ChatMessage) (*pb.SendResponse, error) {
 
 	auth := auth.New(s.pbURL, s.pbAdmin, s.pbPass)
@@ -106,8 +107,7 @@ func (s *Server) Send(ctx context.Context, in *pb.ChatMessage) (*pb.SendResponse
 		return nil, fmt.Errorf("user not authorized")
 	}
 
-	// log.Printf("GRPC: %s/%s->%s", in.UserId, in.ChannelId, in.Text)
-	queue := queue.NewQueue(s.natsURL, "NOT_USED")
+	q := queue.NewQueue(s.natsURL, "")
 	// Override timstamp
 	in.Ts = fmt.Sprint(time.Now().UnixNano())
 
@@ -127,18 +127,18 @@ func (s *Server) Send(ctx context.Context, in *pb.ChatMessage) (*pb.SendResponse
 		}
 	}
 
-	if err := queue.Publish("myserver", payload); err != nil {
+	if err := q.Publish("myserver", payload); err != nil {
 		log.Printf("error publishing message to queue: %v", err)
 		return nil, err
 	}
-	queue.Close()
+	q.Close()
 	return &pb.SendResponse{Ok: true, Error: ""}, nil
 }
 
-func systemMessage(msg string) *pb.ChatMessage {
+func systemMessage(msg, userid string) *pb.ChatMessage {
 	return &pb.ChatMessage{
 		ChannelId: "system", // system information channel - the UI implements behavior based on events received on this channel
-		UserId:    "server", // 'server' is not an actual user
+		UserId:    userid,   // 'server' is not an actual user
 		Text:      msg,
 		Ts:        "0",
 	}
@@ -193,15 +193,19 @@ func (s *Server) getPastMessages(srv pb.ChatService_ConnectServer, in *pb.Connec
 }
 
 // Ping will send a ping message to the client every second and cancel the global context if the client disconnects
-func ping(ctx context.Context, srv pb.ChatService_ConnectServer, cancel context.CancelFunc, streamid string) {
+func (s *Server) ping(ctx context.Context, srv pb.ChatService_ConnectServer, in *pb.ConnectRequest, cancel context.CancelFunc) {
 	for {
 		select {
 		case <-ctx.Done():
+			err := s.broadcast(in.UserId, "disconnected")
+			if err != nil {
+				log.Printf("PING %s: error broadcasting disconnect message: %v", s.streamid, err)
+			}
 			return
 
 		default:
 			time.Sleep(1 * time.Second)
-
+			s.broadcast(in.UserId, "connected")
 			err := srv.Send(&pb.ChatMessage{
 				ChannelId: "system", // system information channel
 				UserId:    "server",
@@ -215,3 +219,8 @@ func ping(ctx context.Context, srv pb.ChatService_ConnectServer, cancel context.
 		}
 	}
 }
+
+// Broadcast sends a message to all connected clients
+func (s *Server) broadcast(user, msg string) error {
+	return s.queue.Publish("myserver", []byte(`{"channelId":"system","userId":"`+user+`","text":"`+msg+`","ts":"0"}`))
+}